阿卡；使用 EventSourcedBehavior 传递太多参数

Question

我们正在构建一个基于 Akka-Typed 的事件源系统。我们很快陷入了这样一种情况，即我们的状态需要许多参数作为隐式参数传递。

样式指南上有一个使用封闭类的解决方案； https://doc.akka.io/docs/akka/current/typed/style-guide.html#passing-around-too-many-parameters.

但是，这是针对“简单”行为，而不是针对 EventSourcedBehaviors。这种使用产生行为的封闭类的技术不适用于事件源行为，因为事件源行为由一个命令处理程序和一个事件处理程序组成。

在我们的例子中，我们有一个State trait，它定义了一种接收命令的方法和另一种处理事件的方法。如果我们应用封闭类技术，我们必须为所有状态创建匿名类，但这些不能被序列化。

object SampleEventSourced {

  trait State extends CborSerializable {
    def execute(cmd: Command): ReplyEffect
    def apply(evt: Event): State
  }

  def apply(
    persistenceId: PersistenceId,
    config: Config,
  ): Behavior[Command] = {
    EventSourcedBehavior
      .withEnforcedReplies[Command, Event, State](
        persistenceId,
        new SampleEventSourced(config).empty(),
        (state, cmd) => state.execute(cmd),
        (state, evt) => state.apply(evt)
      )// ...
  }
}

class SampleEventSourced private(config: Config) {
  import SampleEventSourced._

  private def empty(): State = new State {
    override def execute(cmd: Command): ReplyEffect = cmd match {
      // ..
    }

    override def apply(evt: Event): State = evt match {
      // ..
    }
}

java.lang.IllegalArgumentException：空状态 [org.acme.SampleEventSourced$$anon$1@36d7d2fe] 不可序列化。

一种解决方案是在每个“状态”方法中复制事件源行为的创建。但这会产生很多重复。

object SampleEventSourced {


  def apply(
    persistenceId: PersistenceId,
    config: Config,
  ): Behavior[Command] = new SampleEventSourced(config).empty()
}

class SampleEventSourced private(config: Config) {
  import SampleEventSourced._

  private def empty(): Behavior[Command] = EventSourcedBehavior
      .withEnforcedReplies[Command, Event, State](
        persistenceId,
        new State(),
        (state, cmd) => state.execute(cmd),
        (state, evt) => state.apply(evt)
      )// ...
  }

另一种方法是创建State 的具体子类，但我们必须在所有这些状态之间传递参数。

object SampleEventSourced {

  class EmptyState extends State(config:Config, otherUselessParameter:Any) {
    // ...
    override def apply(evt: Event): evt match {
      case _ => new OtherState(config, otherUselessParameter)
    }
  }
 
  class OtherState extends State(config:Config, veryImportantParameter:Any) {
    // ..
  }
}

将这些状态类放在封闭类中是行不通的，因为这些非静态内部类无法反序列化。

那么，对于这种情况，你的解决方案是什么，你如何处理需要许多参数的状态的 EventSourcedBehavior？

Answer 1

问题在于，jackson-cbor 除了基于反射之外，在序列化以惯用的 Scala 方式完成的事情方面并不是那么有效（它唯一有利的一点是它比 Java 序列化更好）。

因此，显而易见的解决方案是：

• 将状态建模为代数数据类型（通常是由case classes 扩展的sealed trait，对于空/初始状态情况可能是case object）。

• 不要通过 Jackson 序列化它们，而是使用原生到 Scala 序列化器（例如 play-json、circe 等）和 an Akka serializer leveraging that serializer；定义标记特征（例如CirceSerializable）可能更容易，因此在配置中您只需要担心一个绑定。事件和状态将希望使用该特征进行标记：命令及其回复（至少可以通过网络发送的回复也可以）。

定义序列化器时需要注意的一个微妙之处是，原生到 Scala 的序列化器通常会在编译时使用某种形式的类型类派生，但如果需要序列化 ​​ActorRef，这会很复杂，作为ActorRef 的序列化程序，最好在ActorSystem 的帮助下完成（现在可以检查EntityRef 的序列化，并且考虑到生命周期，从概念上讲对于持久性更有意义，但这将是相当定制的）：因此，您可能希望延迟推导 ActorRef 序列化器/反序列化器，直到 ActorSystem 启动，并延迟推导包含 ActorRefs 的事件/状态/命令/回复，直到之后。

我可能倾向于让每个事件的处理程序成为sealed trait 中的一个方法，而不让状态了解事件是如何具体化的。这样做的好处是您可以在不需要事件或命令的情况下对状态转换进行单元测试。

case class ItemAdded(name: String) extends Event

sealed trait State {
  def items: Set[String]

  def addItem(name: String): State.NonEmpty
}

object State {
  case object Empty extends State {
    def items: Set[String] = Set.empty
    def addItem(name: String): NonEmpty = NonEmpty(Set(name))
  }

  case class NonEmpty(items: Set[String]) extends State {
    def addItem(name: String): NonEmpty = copy(items = items + name)
  }
}

那么你的事件处理程序很瘦

(state, evt) => {
  evt match {
    case ItemAdded(name) => state.addItem(name)
  }
}

根据您的命令处理程序进行多少验证，这些可能会变得复杂。操作原则是命令被解释为零个或多个事件，事件被解释为对状态的操作（导致新状态的状态方法）。因此，状态应该有足够的查询方法（例如items）以允许命令处理程序使用状态。