如何保证akka中fork的顺序性答案

【问题标题】：How to guarantee sequentiality for forks in akka如何保证akka中fork的顺序性
【发布时间】：2015-03-18 15:49:20
【问题描述】：

我们正在为每个（小）传入消息组创建一个参与者链，以保证它们的顺序处理和管道（组通过公共 ID 进行区分）。问题是我们的链有分叉，比如A1 -> (A2 -> A3 | A4 -> A5)，我们应该保证通过A2 -> A3 和A4 -> A5 的消息之间没有竞争。当前的遗留解决方案是阻止A1actor 直到当前消息被完全处理（在子链之一中）：

def receive { //pseudocode
    case x => ...
      val f = A2orA4 ? msg
      Await.complete(f, timeout)
}

因此，应用程序中的线程数与正在处理的消息数成正比，无论这些消息是活动的还是只是异步等待外部服务的某些响应。它与 fork-join（或任何其他动态）池一起工作大约两年，但当然不能与固定池一起使用，并且在高负载的情况下会极大地降低性能。不仅如此，它还会影响 GC，因为每个被阻塞的 fork-actor 都在内部保存了冗余的先前消息的状态。

即使有背压，它创建的线程数也比收到的消息多 N 倍（因为流中有 N 个顺序分叉），这仍然很糟糕，因为处理一条消息需要很长时间但 CPU 不多。所以我们应该处理尽可能多的消息，因为我们有足够的内存。我想出的第一个解决方案 - 像A1 -> A2 -> A3 -> A4 -> A5 这样线性化链。有没有更好的？

【问题讨论】：

标签： multithreading scala akka future

【解决方案1】：

更简单的解决方案是将最后收到的消息的未来存储到参与者的状态中，并将其与之前的未来链接：

def receive = process(Future{new Ack}) //completed future
def process(prevAck: Future[Ack]): Receive = { //pseudocode
    case x => ...
        context become process(prevAck.flatMap(_ => A2orA4 ? msg))
}

因此它将创建没有任何阻塞的期货链。该链将在期货完成后被删除（最后一条除外）。

【讨论】：