scala - Akka actor pipeline和拥塞store actor

Question

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

问题是读取、过滤和扩充步骤比存储步骤快得多，这导致存储 actor 拥塞和系统不可靠。

我正在考虑以下选项:让存储 actor 提取已处理并准备存储的消息。这是一个好的选择吗？更好的建议？

谢谢

Answer 1

您可以考虑几种选择:

• 如果消息的顺序无关紧要 - 只需在单独的 actor(或 future)中执行每个存储操作。它将导致所有数据存储并行进行——我建议为此使用单独的线程池。如果某些消息是对其他消息的修改或参与同一事务——您可以仅为每个 messageId/transactionId 创建单独的参与者，以避免悲观/乐观锁问题(不要忘记在事务结束时或超时时杀死此类参与者)。

• 使用有界邮箱(背压)- 如果旧消息仍未处理，您将阻止来自输入的新消息(例如，您可以阻止接收线程，直到消息将被链中的最后一个参与者确认).它将责任转移到源系统。它与 JMS 耐用品一起工作得非常好 - 消息以可靠的方式存储在 JMS 代理端，直到您的系统最终处理它们。

• 结合前两者