2

我正在使用 Akka Cluster 来实现一些冗余并消除我的 java 应用程序中的单点故障。但是，在将消息从一台机器发送到另一台机器时，我在消息的序列化方面遇到了一些问题。

我正在发送一个未实现的 java 对象Serializable。因此，我需要使用自己的序列化程序，幸运的是，该对象提供了使用 Kryo 的序列化方法。因此，按照本指南，我创建了自己的 akka 序列化程序扩展JSerializer：

import akka.serialization.JSerializer;

public class MyObjectClusterSerializer extends JSerializer {

    public int identifier() {
        return 66666;
    }

    public boolean includeManifest() {
        return false;
    }

    public byte[] toBinary(Object obj) {
        MyObject myObject = (MyObject) obj;
        byte[] bytes = null;
        try {
            bytes = MyObjectHelper.serialize(myObject);
        } catch (SerializationException e) {

        }
        return bytes;
    }

    @Override
    public Object fromBinaryJava(byte[] arg0, Class<?> arg1) {
        MyObject myObject = null;
        try {
            myObject = MyObjectHelper.deserialize(arg0);
        } catch (DeserializationException e) {
        }
        return myObject;
    }

}

代码非常简单，我不认为这是问题所在。然后我将序列化程序和序列化程序绑定添加到我的配置中：

serializers {
  myobject = "com.my.awesome.project.MyObjectClusterSerializer"
}

serialization-bindings {
  "com.my.awesome.project.MyObject" = myobject
}

当我运行单个节点时，我没有任何问题（显然是因为我没有序列化任何东西......）。但是在集群模式下，我不断收到这些异常：

    2014-08-04 01:45:27,498 ERROR [MyApp-akka.actor.default-dispatcher-16] - Slf4jLogger$$anonfun$receive$1$$anonfun$applyOrElse$1.apply$mcV$sp (Slf4jLogger.scala:66) - Transient association error (association remains live)
    java.io.NotSerializableException: com.my.awesome.project.MyObject
            at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1183) ~[?:1.7.0_60]
            at java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1547) ~[?:1.7.0_60]
            at java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1508) ~[?:1.7.0_60]
            at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1431) ~[?:1.7.0_60]
            at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1177) ~[?:1.7.0_60]
            at java.io.ObjectOutputStream.writeObject(ObjectOutputStream.java:347) ~[?:1.7.0_60]
            at akka.serialization.JavaSerializer$$anonfun$toBinary$1.apply$mcV$sp(Serializer.scala:129) ~[akka-actor_2.11-2.3.4.jar:?]
            at akka.serialization.JavaSerializer$$anonfun$toBinary$1.apply(Serializer.scala:129) ~[akka-actor_2.11-2.3.4.jar:?]
            at akka.serialization.JavaSerializer$$anonfun$toBinary$1.apply(Serializer.scala:129) ~[akka-actor_2.11-2.3.4.jar:?]
            at scala.util.DynamicVariable.withValue(DynamicVariable.scala:58) ~[scala-library-2.11.1.jar:?]
            at akka.serialization.JavaSerializer.toBinary(Serializer.scala:129) ~[akka-actor_2.11-2.3.4.jar:?]
            at akka.remote.serialization.MessageContainerSerializer.serializeSelection(MessageContainerSerializer.scala:36) ~[akka-remote_2.11-2.3.4.jar:?]
            at akka.remote.serialization.MessageContainerSerializer.toBinary(MessageContainerSerializer.scala:25) ~[akka-remote_2.11-2.3.4.jar:?]
            at akka.remote.MessageSerializer$.serialize(MessageSerializer.scala:36) ~[akka-remote_2.11-2.3.4.jar:?]
            at akka.remote.EndpointWriter$$anonfun$serializeMessage$1.apply(Endpoint.scala:845) ~[akka-remote_2.11-2.3.4.jar:?]
            at akka.remote.EndpointWriter$$anonfun$serializeMessage$1.apply(Endpoint.scala:845) ~[akka-remote_2.11-2.3.4.jar:?]
            at scala.util.DynamicVariable.withValue(DynamicVariable.scala:58) ~[scala-library-2.11.1.jar:?]
            at akka.remote.EndpointWriter.serializeMessage(Endpoint.scala:844) ~[akka-remote_2.11-2.3.4.jar:?]
            at akka.remote.EndpointWriter.writeSend(Endpoint.scala:747) ~[akka-remote_2.11-2.3.4.jar:?]
            at akka.remote.EndpointWriter$$anonfun$4.applyOrElse(Endpoint.scala:722) ~[akka-remote_2.11-2.3.4.jar:?]
            at akka.actor.Actor$class.aroundReceive(Actor.scala:465) ~[akka-actor_2.11-2.3.4.jar:?]
            at akka.remote.EndpointActor.aroundReceive(Endpoint.scala:415) ~[akka-remote_2.11-2.3.4.jar:?]
            at akka.actor.ActorCell.receiveMessage(ActorCell.scala:516) [akka-actor_2.11-2.3.4.jar:?]
            at akka.actor.ActorCell.invoke(ActorCell.scala:487) [akka-actor_2.11-2.3.4.jar:?]
            at akka.dispatch.Mailbox.processMailbox(Mailbox.scala:238) [akka-actor_2.11-2.3.4.jar:?]
            at akka.dispatch.Mailbox.run(Mailbox.scala:220) [akka-actor_2.11-2.3.4.jar:?]
            at akka.dispatch.ForkJoinExecutorConfigurator$AkkaForkJoinTask.exec(AbstractDispatcher.scala:393) [akka-actor_2.11-2.3.4.jar:?]
            at scala.concurrent.forkjoin.ForkJoinTask.doExec(ForkJoinTask.java:260) [scala-library-2.11.1.jar:?]
            at scala.concurrent.forkjoin.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1339) [scala-library-2.11.1.jar:?]
            at scala.concurrent.forkjoin.ForkJoinPool.runWorker(ForkJoinPool.java:1979) [scala-library-2.11.1.jar:?]
            at scala.concurrent.forkjoin.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:107) [scala-library-2.11.1.jar:?]

对我来说，这意味着它无法弄清楚要使用什么序列化程序，它实际上使用的是通用 Java 序列化程序。另外，我不知道它是否相关，但我用 ConsistentHashableEnvelope 包装我的消息，因为我将它发送到集群感知的一致哈希组。

有什么帮助可以使这件事起作用吗？我错过了什么吗？

4

1 回答 1

3

Akka 按消息类选择一次序列化程序（参见堆栈跟踪）。在您的情况下，akka 正确使用JavaSerializer因为基本消息具有 ConsistentHashableEnvelope 类型并将此序列化程序应用于所有字段（message和hashKey）。

要解决您的任务，只需为ConsistentHashableEnvelope创建序列化程序并仅在消息字段具有MyObject类型和JavaSerializer在其他情况下使用MyObjectClusterSerializer

于 2014-12-03T10:51:29.050 回答