java - 对 Akka UntypedPersistentActor 的同步(或 future 返回)调用？

Question

我想通知调用者已成功应用发送到我的对象的状态更改命令，它扩展了 UntypedPersistentActor。

如果我不持久化状态更改，我会返回一个带有影响更改的闭包的 Future，并让调用者等待它。

Akka 的“persist()”调用迫使我传入一个将异步执行的闭包，如果我理解正确的话。如果我返回一个本身调用 persist() 的 Future，我只是告诉调用者我已经成功地将更改排队等待以后的应用……而不是它已经被应用，甚至它的应用将会成功。

我想关于一个调用者 kinda 的保证顺序满足了知道何时应用更改的需要，但如果更改失败怎么办？如果被调用者因为失败而被重启，调用者的消息将被丢弃，状态改变将永远不会发生，调用者将毫无察觉。似乎将错误返回给调用者并在那里显示它会更清晰。

有什么好的方法可以实现吗？

Answer 1

不要从 Future 中调用 persist。它必须在接收方的上下文(线程)中调用，因为它依赖于参与者内部状态来进行内务处理。

最简单的方法是使用来自 akka.pattern.ask 的 ask 模式:

case c: Command =>
  persist(Event(c)) { sender() ! e }

// caller:
import akka.pattern.ask
val f: Future[Event] = (persistentActor ? Command()).mapTo[Event]

您还可以使用一个 Promise[T]，您的 actor 可以实现它，并将该 promise 的 Future[T] 发送回发送者:

case c: Command =>
  val p = Promise[Event]()

  persist(Event(c)) { e => 
    p success e 
  }

  sender() ! p.future // Future[T]