使用 Akka 持久邮箱进行事务性消息处理答案

【问题标题】：Transactional message processing with an Akka durable mailbox使用 Akka 持久邮箱进行事务性消息处理
【发布时间】：2015-11-27 10:32:08
【问题描述】：

我有一个与持久性JMS (ActiveMQ) 队列集成的 Java/Akka 应用程序。

PersistentQueue 包装了一个包含批处理作业的 JMS/ActiveMQ 队列。在事务中接收消息，因此如果服务器在作业执行过程中出现故障，那么该作业将在重新启动时保留。如果作业成功完成或被用户取消，则提交此事务以永久删除消息，如果作业失败，则如果作业执行较少，则事务回滚（将消息放在队列的前面）超过 MAX_RETRY 次。

BatchManager 是与 REST 控制器的接口。由于作业执行期间调用的存储过程的限制，它一次只能执行一个批处理作业。 BatchManager 从控制器接收作业并将它们发送到 PersistentQueue 以放入 JMS 队列中，然后在作业入队时轮询 PersistentQueue 以获取新作业（除非另一个作业正在执行）或当工作完成。

我想删除 JMS 队列以及处理其JMSExceptions 的所有复杂性，并将其替换为BatchManager 的持久邮箱。问题是我不知道如何使用持久邮箱复制 JMS 事务 - 我的理解是，如果服务器在作业执行期间出现故障，那么该消息将永远丢失（而不是被放回队列中） JMS 队列）。

有没有办法使用 Akka 持久邮箱来实现事务性消息处理，这样如果服务器在执行过程中出现故障，消息就不会丢失？

【问题讨论】：

标签： java transactions jms akka message-queue

【解决方案1】：

Akka documentation 说：

持久邮箱与任何其他不太可能具有事务性的邮箱一样。如果 Actor 在收到消息后但在完成处理之前崩溃，则消息可能会丢失。

但还有另一种类型的邮箱 - 带有显式确认的邮箱（又名 PeekMailbox）。 Here 你可以找到使用示例。而here是实现源码。

我认为您可以通过实现自定义持久邮箱来实现您的目标，该邮箱扩展了一些现有实现并使用 PeekMailbox 功能对其进行了扩充。

【讨论】：

【解决方案2】：

您可以使用PersistentActor 来跟踪已发布和已完成的工作。

“分布式工作人员”激活器模板包含这样一个工作单元管理参与者。（以及动态工作器注册和集群，但即使您不感兴趣，它仍然值得一看）。

Scala version of the template 和 Java version of the template

【讨论】：