已完成所有 Scala 期货

Question

我有一个 akka 演员，其中我在 scala 中有几个期货，它们只做一些额外的工作并将计算保存到共享变量中。主要工作完成后，我想等待这些“额外”的未来并回复一个答案。简单的代码是这样的：

....
val replyTo = sender()
var informationCollector = InformationCollector()
val mainFuture: Future[GetMainInformation] = ...
val aFuture: Future[GetExtraInformationA] =...
val bFuture: Future[GetExtraInformationB] =...
val cFuture: Future[GetExtraInformationC] =...

// now I want all the futures to write their result
// and save it into the informationCollector and then
// after all have executed, reply back to Sender all the collected Data.

使用这种方法，我有 2 个问题。我不确定在期货的 onSuccess 调用中访问 InformationCollector 是否安全，我也想让它易于扩展（因此可以轻松添加其他收集器）。

到目前为止，我想出了这个解决方案，但我不确定它是否正确

// IMPORTANT: firstly add map for all of the extraInfoFutures to add that information into collector
// this will create aFutureMapped, bFutureMapped, cFutureMapped
roomFuture onComplete {
      case Success(mainInfo: GetMainInformation) => {
           informationCollector = informationCollector.copy(mainInformation=mainInfo)
           Future.sequence(List(aFutureMapped,bFutureMapped,cFutureMapped)) onComplete { _ =>
               replyTo ! informationCollector
           }

      }
}

非常感谢任何帮助。

Answer 1

我不确定我是否完全理解您的问题，但我至少可以为第一个问题提供答案。

访问onComplete 中的可变状态可能是不安全的。实际上，它使演员的全部目的无效。如果您访问可变状态，您应该始终通过参与者的邮箱进行访问，例如self ! ModifyState(newValue).

这里的问题是onComplete 方法与参与者不在同一个线程中，这导致两个线程可能同时修改相同数据并因此改变状态具有与通常并发的所有相同问题状态突变（竞争条件等）。

对于第二个问题，我将通过self ! TypeAExtraValue(someValue) 从不同类型的额外期货中收集值，然后在收到所有额外值后发送收集器（例如TypeAInfoCollector）。如果使用得当，actor 可以解决这个问题，因为它们一次只处理一条消息。

代码可能如下所示：

val extraInfoFuture = getExtraInfo()
val infoCollector = new Collector()

extraInfoFuture onComplete {
  case Success(extraInfo) => self ! ExtraInfo(extraInfo) 
}

...

def receive = {
  case ExtraInfo(info) => {
    infoCollector.collect(info)
    if(infoCollector.collectedAll) sender() ! infoCollector
  }
}