无法使用 jackson-cbor 在 Akka 中序列化密封特征

Question

我正在尝试调整Akka distrubuted workers sample 项目。基本上，我想创建更多的Work项目，这些项目将从FrontEnd发送，然后由WorkExecutor以分布式方式计算，然后将结果返回给FrontEnd。

我的问题是我想扩展 Work 以一种我可以通过其他类型的工作（扩展 Work）以相当通用的方式（例如，WorkManager 应该只处理 @987654329 @)。

sealed trait Work extends CborSerializable {
  def workId: String

  def job: Int
}

case class MyWorkItem(workId: String, job: Int) extends Work

很遗憾，jackson-cbor 无法序列化Work。

4163 [ClusterSystem-akka.actor.default-dispatcher-5] ERROR org.test.worker.WorkManager$ - Supervisor StopSupervisor saw failure: Exception during recovery. Last known sequence number [2]. PersistenceId [master], due to: Exception during recovery. Last known sequence number [2]. PersistenceId [master], due to: Cannot construct instance of `org.test.worker.Work` (no Creators, like default construct, exist): abstract types either need to be mapped to concrete types, have custom deserializer, or contain additional type information
 at [Source: (byte[])"�dwork�fworkIdx$f031c8c6-6b7e-4e63-9beb-e97330eea35ecjob��"; line: -1, column: 6] (through reference chain: org.test.worker.WorkState$WorkAccepted["work"])
akka.persistence.typed.internal.JournalFailureException: Exception during recovery. Last known sequence number [2]. PersistenceId [master], due to: Exception during recovery. Last known sequence number [2]. PersistenceId [master], due to: Cannot construct instance of `org.test.worker.Work` (no Creators, like default construct, exist): abstract types either need to be mapped to concrete types, have custom deserializer, or contain additional type information
 at [Source: (byte[])"�dwork�fworkIdx$f031c8c6-6b7e-4e63-9beb-e97330eea35ecjob��"; line: -1, column: 6] (through reference chain: org.test.worker.WorkState$WorkAccepted["work"])
    at akka.persistence.typed.internal.ReplayingEvents.onRecoveryFailure(ReplayingEvents.scala:220)
    at akka.persistence.typed.internal.ReplayingEvents.onJournalResponse(ReplayingEvents.scala:153)
    at akka.persistence.typed.internal.ReplayingEvents.onMessage(ReplayingEvents.scala:91)
    at akka.persistence.typed.internal.ReplayingEvents.onMessage(ReplayingEvents.scala:66)
    at akka.actor.typed.scaladsl.AbstractBehavior.receive(AbstractBehavior.scala:84)
    at akka.actor.typed.Behavior$.interpret(Behavior.scala:274)
    at akka.actor.typed.Behavior$.interpretMessage(Behavior.scala:230)
    at akka.actor.typed.internal.InterceptorImpl$$anon$2.apply(InterceptorImpl.scala:57)
    at akka.persistence.typed.internal.EventSourcedBehaviorImpl$$anon$1.aroundReceive(EventSourcedBehaviorImpl.scala:171)
    at akka.actor.typed.internal.InterceptorImpl.receive(InterceptorImpl.scala:85)
    at akka.actor.typed.Behavior$.interpret(Behavior.scala:274)
    at akka.actor.typed.Behavior$.interpretMessage(Behavior.scala:230)
    at akka.actor.typed.internal.InterceptorImpl$$anon$2.apply(InterceptorImpl.scala:57)
    at akka.actor.typed.internal.SimpleSupervisor.aroundReceive(Supervision.scala:124)
    at akka.actor.typed.internal.InterceptorImpl.receive(InterceptorImpl.scala:85)
    at akka.actor.typed.Behavior$.interpret(Behavior.scala:274)
    at akka.actor.typed.Behavior$.interpretMessage(Behavior.scala:230)
    at akka.actor.typed.internal.InterceptorImpl$$anon$2.apply(InterceptorImpl.scala:57)
    at akka.actor.typed.internal.InterceptorImpl.receive(InterceptorImpl.scala:87)
    at akka.actor.typed.Behavior$.interpret(Behavior.scala:274)
    at akka.actor.typed.Behavior$.interpretMessage(Behavior.scala:230)
    at akka.actor.typed.internal.InterceptorImpl$$anon$2.apply(InterceptorImpl.scala:57)
    at akka.actor.typed.internal.SimpleSupervisor.aroundReceive(Supervision.scala:124)
    at akka.actor.typed.internal.InterceptorImpl.receive(InterceptorImpl.scala:85)
    at akka.actor.typed.Behavior$.interpret(Behavior.scala:274)
    at akka.actor.typed.Behavior$.interpretMessage(Behavior.scala:230)
    at akka.actor.typed.internal.adapter.ActorAdapter.handleMessage(ActorAdapter.scala:129)
    at akka.actor.typed.internal.adapter.ActorAdapter.aroundReceive(ActorAdapter.scala:106)
    at akka.actor.ActorCell.receiveMessage(ActorCell.scala:577)
    at akka.actor.ActorCell.invoke(ActorCell.scala:547)
    at akka.dispatch.Mailbox.processMailbox(Mailbox.scala:270)
    at akka.dispatch.Mailbox.run(Mailbox.scala:231)
    at akka.dispatch.Mailbox.exec(Mailbox.scala:243)
    at java.base/java.util.concurrent.ForkJoinTask.doExec(ForkJoinTask.java:290)
    at java.base/java.util.concurrent.ForkJoinPool$WorkQueue.topLevelExec(ForkJoinPool.java:1020)
    at java.base/java.util.concurrent.ForkJoinPool.scan(ForkJoinPool.java:1656)
    at java.base/java.util.concurrent.ForkJoinPool.runWorker(ForkJoinPool.java:1594)
    at java.base/java.util.concurrent.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:177)

Answer 1

错误是

无法构造org.test.worker.Work的实例

这是因为Work 是一个抽象类，因此无法构造。

答案也在错误信息中：

抽象类型要么需要映射到具体类型，要么需要自定义反序列化器，要么包含额外的类型信息

请注意，它是 sealed 的事实并不相关，问题是它是 trait 而不是具体的 class。

Answer 2

TL/DR

final case class Zoo(primaryAttraction: Animal) extends MySerializable

@JsonTypeInfo(use = JsonTypeInfo.Id.NAME, property = "type")
@JsonSubTypes(
  Array(
    new JsonSubTypes.Type(value = classOf[Lion], name = "lion"),
    new JsonSubTypes.Type(value = classOf[Elephant], name = "elephant")))
sealed trait Animal

final case class Lion(name: String) extends Animal

final case class Elephant(name: String, age: Int) extends Animal

以上是此链接https://doc.akka.io/docs/akka/current/serialization-jackson.html#polymorphic-types的复制粘贴

说明

假设您有一个在您的情况下接受 Work 实例的方法，但问题是您可能有许多 Work 特征的实现，所以 Akka（或者更确切地说是杰克逊）不会知道哪个实现Work 它需要映射传入的 JSON。通过使用JsonTypeInfo 和JsonSubTypes，您指示Jackson 将type 属性添加到序列化的JSON，然后将用于将JSON 反序列化回您的具体类。

您可能会争辩说您的 trait 是密封的，Jackson/Akka 可能会发现您只有一个实现并使用它，但我想这没有实现并且可能不太容易实现。

这是一个棘手的问题，因为此错误只会在消息从集群中的一个节点发送到另一个节点时出现，并且如果消息在同一节点上的参与者之间发送则不会出现。