Я использую кластер Akka для достижения некоторой избыточности и удаления отдельных точек отказа в моем приложении Java. Однако у меня возникают некоторые проблемы с сериализацией сообщений при отправке их с одной машины на другую.Akka Cluster custom serializer:
Я отправляю объект java, который не осуществляет Serializable. Из-за этого мне нужно использовать собственный сериализатор, и, к счастью, объект предоставляет метод сериализации с использованием Kryo. Итак, после этого guide я создал свой собственный 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, потому что я отправляю его в группу хэширования, поддерживающую кластер.
Любая помощь, чтобы эта работа работала? Я что-то упускаю?
Подобно тому, как это не ясно из вашей вставленной конфигурации, у вас есть сериализаторы и привязки сериализации под akka.actor в вашей конфигурации, правильно? – pushy
Привет, настойчивый, да, это под аккой. – hveiga