RabbitMQ - 每个用户只允许一个进程答案

【问题标题】：RabbitMQ - allow only one process per userRabbitMQ - 每个用户只允许一个进程
【发布时间】：2020-03-01 06:33:34
【问题描述】：

为了简短起见，这里有一个简化的情况：

我需要为导入的数据文件的后台处理实现一个队列。我想为这个特定任务指定一些消费者（比如说 10 个），以便可以并行处理多个用户。同时，为了避免并发数据写入的问题，我需要确保没有一个用户同时在多个消费者中处理，基本上单个用户的所有文件都应该顺序处理。

目前的解决方案（但感觉不对）：

有 1 个队列用于发布所有导入任务 (file_queue_main)
有 10 个队列用于文件处理 (file_processing_n)
有 1 个结果队列 (file_results_queue)
有一个管理器进程（在本例中为node.js），它会一个一个地使用来自file_queue_main 的消息，并决定将该消息分发到哪个file_processing 队列。基本上跟踪当前用户正在处理的file_processing 队列。

这是我当前解决方案和预期行为的小动画：

RabbitMQ 甚至是这项工作的工具吗？出于某种原因，它感觉像是某种反模式。感谢任何帮助！

【问题讨论】：

由于您希望为每个用户序列化文件，因此您需要有用户特定的队列。然后可以根据负载和流量条件生成消费者。
感谢您的评论。据我了解，这也意味着每个用户都会有一个消费者？如何将消费者数量限制为 10，同时确保一些随机的 10 个用户不会填满队列，使其他用户永远等待？在我的解决方案中，消息按照它们进入的顺序进行处理，但是如果每个用户都有一个队列，那么当用户开始处理时，只要消费者已经启动，他就可以无限地添加文件。
我认为 Kafka 可能是一个更清洁的解决方案。您可以绕过竞争工人模式的所有协调和复杂性。使用 Kafka 来实现并行性并提高吞吐量，您只需向 Kafka 主题添加更多代理，就是这样！
也许我遗漏了一些东西，但是如果您创建十个队列（和十个消费者），然后在您将用户发布到交易所时分配具有等于user_id mod 10 的路由密钥的用户。然后你可以确定同一个用户总是进入同一个队列。即用户120去queue_0，121去queue_1...用户123451去queue_1等等？
目标是有效利用所有 10 个消费者。如果我通过简单的mod 10 分发消息，那么很有可能一些消费者会空闲。

标签： rabbitmq queue sharding

【解决方案1】：

对我来说“感觉不对”的部分是经理流程。它必须知道每个消费者的当前状态，如果所有处理器都在为其他用户工作，它还必须停止并等待。理想情况下，您更愿意让每个进程不了解其他进程。您从处理队列中获得的好处也很少，只有当处理器已经在处理来自同一用户的消息时才会使用这些队列。

最终，这里的最佳解决方案将取决于您的预期使用情况以及下一条消息来自已被处理的用户的可能性有多大。如果您希望在任何时候收到的大多数消息都来自 10 个或更少的用户，那么您所拥有的可能没问题。如果您希望处理来自许多不同用户的消息，并且偶尔会出现重复消息，那么您的处理队列大部分时间都会是空的，并且会产生很多不必要的复杂性。

您可以在这里做的其他事情：

让所有消费者从同一个队列中拉出并使用某种分布式锁定来防止冲突。如果消费者从已经在处理的用户那里收到消息，请将其重新排队并继续。
设置您的队列路由，以便来自同一用户的消息始终发送到同一消费者。不利的一面是，如果您不将流量平均分配出去，您可能会让一些消费者备份，而另一些消费者则闲置。

另外，如果您一次从同一用户那里收到大量必须按顺序处理的消息，我会质疑它们是否应该是单独的消息。为什么不发送一条包含要处理的事项列表的消息？事件队列的大部分好处在于能够将每个事件视为可以单独处理的离散项。

【讨论】：

主要目标是有效利用所有 10 个消费者。如果我按模式路由消息，那么大多数消费者很可能会闲置而不使用。无法预测每个用户的文件有多少或有多大。基本上，我想为每个需要做一些繁重工作的用户创建一个队列。

【解决方案2】：

如果用户具有唯一 ID，或者正在处理的文件具有唯一 ID，则对 ID 进行哈希处理以获取要进入的处理队列。这样，您将始终有相同的用户/文件任务在同一个处理队列中排队。

我不确定这将如何影响处理队列的队列长度。

【讨论】：

这就是我想要避免的。无法预测用户的活跃程度。所以我需要能够将任何用户分配给第一个可用的消费者。