如何与actor实现异步相互依赖的可取消操作？

Question

我对 Actor 模型还很陌生，这就是为什么我认为已经有既定的模式可以用 Actor 和 future 等美丽的可组合抽象来解决我的常见场景。

我的异步操作具有以下要求：

• 它们通过发送低级请求然后通过轮询监视实体的状态来使用遗留系统。因此，实际操作的结果只能以延迟的方式获得，当观察到的状态达到所需状态时，必须通知请求者。
• 这些操作只有在其他一些操作完成后才能发出，并且它们应该并行等待。
• 可以取消操作。当然，已经发出的低级请求无法撤消；取消意味着在我们依赖的操作完成后不发出实际操作，当然这必须递归传播（如果我们等待依赖项，并且它有多个挂起的操作，则不要发出它们）。

我在Futures中思考：第一个要求可以用Akka的map/flatMap来解决，第二个要求可以用traverse组合器来解决，而无需维护依赖关系/程序上的依赖者。但我想不出取消的解决方案； futures 不能被取消，如果组合，它们的组件也无法访问。如何以函数式方式封装“取消当前操作”？ Scala 的 Actor 框架是否支持此功能？

Answer 1

使用监听器： https://github.com/jboner/akka/blob/release-1.2/akka-actor/src/main/scala/akka/routing/Listeners.scala

制作一个使用的 Actor侦听器将轮询状态传播给所有侦听器。然后您可以使用消息传递循环来重新启动轮询。

class MyActor extends Actor with Listeners {

  override def preStart {
    self ! 'poll //Start looping on start
  }

  def receive = listenerManagement orElse {
    case 'poll => val result = pollYourExternalDude()
    gossip(result)
    self ! 'poll //Loop
  }
}

然后，您可以使用 stop 或发送 PoisonPill 来停止 actor。

这有帮助吗？

Answer 2

Guava 的 ListenableFuture支持取消到 绑定在一起（但当 从集合中组合）。