Akka 演员管道和拥挤的商店演员答案

【问题标题】：Akka actor pipeline and congested store actorAkka 演员管道和拥挤的商店演员
【发布时间】：2017-06-20 15:39:38
【问题描述】：

我正在尝试使用参与者实现消息处理管道。管道的步骤包括读取、过滤、扩充以及最后存储到数据库等功能。类似的东西：http://sujitpal.blogspot.nl/2013/12/akka-content-ingestion-pipeline-part-i.html

问题在于读取、过滤和扩充步骤比存储步骤快得多，这会导致存储参与者拥塞和系统不可靠。

我正在考虑以下选项：让 store actor 拉出已处理并准备好存储消息。这是一个不错的选择吗？更好的建议？

谢谢

【问题讨论】：

标签： scala akka actor

【解决方案1】：

您可以考虑几种选择：

如果消息的顺序无关紧要 - 只需在单独的参与者（或未来）内执行每个存储操作。这将导致所有数据存储并行进行 - 我建议为此使用单独的线程池。如果某些消息是对其他消息的修改或参与同一事务 - 您可以仅为每个 messageId/transactionId 创建单独的参与者以避免悲观/乐观锁定问题（不要忘记在事务结束或超时时杀死这些参与者）。
使用有界邮箱（背压） - 如果旧消息仍未处理，您将阻止来自输入的新消息（例如，您可能会阻止接收线程，直到消息将被链中的最后一个参与者确认）。它将责任转移到源系统。它与 JMS 持久性数据配合得非常好 - 消息以可靠的方式存储在 JMS 代理端，直到您的系统最终处理它们。
结合前两个

【讨论】：

第一个选项考虑使用路由器。
我相信你的第一个选择是我现在正在做什么？ IE。我有一个商店演员。由于 mongo 中的整个写锁定问题，我无法拥有更多。我会看看第二个选项。
是的，第一个选项不适用于 mongo，除非您有多个数据库要写入（自 2.2 以来，mongo 将仅锁定数据库而不是整个服务器） - 然后您可以为每个数据库创建写入 actor。
最终按照您的建议将两者结合起来，并尽可能将我的数据拆分到多个数据库中。效果很好。

【解决方案2】：

我正在使用与此类似的方法：Akka Work Pulling Pattern（此处为源代码：WorkPullingPattern.scala）。它的优点是它可以在本地和 Akka Cluster 中工作。此外，整个方法完全异步，完全没有阻塞。

如果您处理的“对象”不能全部放入内存，或者其中一个步骤很慢，那么这是一个很棒的解决方案。如果您产生 N 个工作人员，那么一次将处理 N 个“任务”。将“步骤”放入BalancingPools 可能是一个好主意，同时并行度为 N（或更少）。

我不知道您的处理“管道”是否是顺序的，但如果是，就在几个小时前，我已经开发了一个基于上述 + Shapeless 库的类型安全抽象。在与 WorkPullingPattern 合并之前的代码一览如下：Pipeline。

它采用任何函数管道（具有正确匹配的签名），在 BalancingPools 中生成它们，创建 Worker 并将它们链接到可用于调度任务的主参与者。

【讨论】：

我喜欢这个解决方案，但是其他答案建议意味着对我的代码的更改更少并且最终工作得很好，所以我将其标记为答案。

【解决方案3】：

新的 AKKA 流（仍处于测试阶段）具有背压。它旨在解决这个问题。

【讨论】：

【解决方案4】：

你也可以在actors上使用receive pipeline：

class PipelinedActor extends Actor with ReceivePipeline {

  // Increment
  pipelineInner { case i: Int ⇒ Inner(i + 1) }
  // Double
  pipelineInner { case i: Int ⇒ Inner(i * 2) }

  def receive: Receive = { case any ⇒ println(any) }
}

actor ! 5 // prints 12 = (5 + 1) * 2

http://doc.akka.io/docs/akka/2.4/contrib/receive-pipeline.html

它最适合您的需求，因为您在演员处理消息之前/之后有小型管道任务。它也是阻塞代码，但我相信这对你的情况很好

【讨论】：