在Akka演员中使用Futures

Question

我刚刚开始学习Scala中的Akka Actors.我的理解是,Actor接收的消息在Actor的邮箱中排队,并且一次处理一个.通过一次处理一个消息,可以减轻并发问题(竞争条件,死锁).

但是如果Actor创建了一个与消息相关的工作的未来会发生什么？由于未来是异步的,因此Actor可以开始处理接下来的几条消息,而与先前消息相关联的未来仍然在运行.这不会产生竞争条件吗？如何在Actor的receive()方法中安全地使用future来执行长时间运行的任务？

Answer 1

在演员中使用期货最安全的方法是只使用pipeTo未来并将其结果作为消息发送给演员(可能是同一个演员).

import akka.pattern.pipe

object MyActor {

  def doItAsynchronously(implicit ec: ExecutionContext): Future[DoItResult] = { 
  /* ... */ 
  }

}

class MyActor extends Actor {

  import MyActor._    
  import context.dispatcher  

  def receive = {
    case DoIt =>
      doItAsynchronously.pipeTo(self)
    case DoItResult =>
       // Got a result from doing it
  }

}

这可以确保您不会改变actor中的任何状态.

Answer 2

如果您需要在未来改变状态而不阻止传入消息，您可能需要重新考虑重新设计您的参与者模型。我将为您将使用这些 future 的每项任务引入单独的参与者。毕竟，Actor 的主要任务是维持其状态而不让它逃逸，从而提供安全的并发性。为那些长期运行的任务定义一个参与者，其职责只是处理该任务。

您可能需要考虑使用akka 的 FSM，而不是手动处理状态，这样您就可以更清晰地了解何时发生变化。当我处理更复杂的系统时，我个人更喜欢这种方法而不是丑陋的变量。