ActiveMQ：如何分叉加入？ IE如何在所有子任务完成后发出一条消息

如何解决ActiveMQ：如何分叉加入？ IE如何在所有子任务完成后发出一条消息

假设你有一些任务结构

Task1
Task2: 1 million separate independent Subtask[i] that can run concurrently
Task3: must run once after ALL Task2 subtasks have completed

而Task1、Subtask[i]和Task3都是用MQ消息来表示的。 如何在 ActiveMQ 上解决这个问题？尤其是在所有子任务完成后触发 Task3 消息。

我知道，这不是排队问题，而是 fork-join 问题。假设环境要求您必须为此使用 ActiveMQ。

允许使用 ActiveMQ 功能、动态队列和消费者等。不允许使用外部计数器，例如代表 Task2 进度的数据库行。

解决方法

隐藏在这个 fork-join 问题中的是状态管理和可观察性挑战。由于排除了数据库，您必须依赖内存中或队列中的内容。

1. 为任务运行创建一个唯一的 id —— 很短，但有足够的空间不会像飞机定位器代码那样发生碰撞 —— 即。 34FDSX
2. 将任务的所有消息发送到队列://TASK.34FDSX.DATA
3. 向 queue://TASK.34FDSX.CONTROL 发送控制消息，其中包含任务 ID 和预期的消息总数（包括每个 messageId 也会有帮助）
4. 当 queue://TASK.34FDSX.DATA 中的消费者完成他们的工作时，他们应该向 queue://TASK.34FDSX.DONE 队列发送一条“完成”消息，其中包含他们的 messageId 或一些标识符。

.CONTROL 队列和 .DONE 队列的消费者应该是同一个进程，并且可以跟踪预期的和总完成的任务。一切都完成后，他可以触发事件以触发任务 #3。

这种方法将所有内容都提供为“在线”，如果在任务完成前的时间过长，您还可以使 .CONTROL 和 .DONE 读取器超时。

队列删除可以使用 ActiveMQ 目标 GC 来完成，也可以在一切成功完成时作为 .CONTROL/.DONE 读取器中的清理步骤来完成。

优点：

• 没有无限阻塞消费者
• 没有无限的未结交易
• 任务的状态是在线的，并且可以通过队列和队列指标（队列大小、入队计数、出队计数）的存在进行观察
• 整个解决方案可以是多线程的，唯一的要求是对于给定的任务，.CONTROL/.DONE 侦听器是同一个使用者，但多个任务可以有单独的 .CONTROL/.DONE 侦听器进行扩展。立>

,

这里的问题有点含糊，所以我的回答也必须有点含糊。

“任务 2”的​​百万个独立子任务中的每一个都可以由一条消息表示。所有这些消息都可以在同一个队列中。您可以根据需要启动任意数量的消费者并处理所有这些消息（即执行所有子任务）。只需确保这些消费者使用客户端确认模式或事务处理会话，以便在完成处理消息之前不会将消息从队列中删除。一旦队列中没有更多消息，您就知道“任务 2”已完成。

要检测队列何时为空，您可以在队列上设置一个“特殊”使用者，该使用者会定期打开事务会话并尝试使用队列中的消息。如果消费者收到一条消息，那么您可以回滚事务会话以将消息放回队列，并且您知道队列不为空（即“任务 2”未完成）。如果消费者没有收到消息，那么您就知道队列是空的，您可以发送另一条消息来表明这一点。您可以在发送完子任务的所有消息后，将这个特殊使用者作为“任务 2”的​​一部分启动，以避免过早检测到空队列。

明确地说，这是一个简单的解决方案。您当然可以根据您的要求增加更多复杂性，但您的问题只是概述了基本问题，因此不清楚您还有哪些其他要求（如果有）。