如何基于标准限制并发消息消费

Question

场景(我已经简化了事情)：

• 许多最终用户可以从前端web应用程序(生产者)开始作业(例如，繁重的任务，比如生成一个大的PDF )。
• 这些作业被发送到一个持久的RabbitMQ队列。
• 许多工人应用程序(消费者)处理这些作业并将结果写入数据存储中。

这个相当标准的模式运行良好。

The problem：如果一个用户在同一分钟内启动了10个作业，而在一天中的那个时候只有10个工人应用程序在运行，那么这个最终用户实际上正在为自己接管所有的计算时间。

问题：我如何确保每个最终用户在任何时候只处理一个作业？(奖励：某些最终用户(例如，管理员)不能被限制)

此外，我不希望前端应用程序阻止最终用户启动并发作业。我只想让最终用户等待他们的并发作业一次完成一次。

的解决方案？：我应该动态地为每个终端用户创建一个自动删除的排他队列吗？如果是，我如何告诉辅助应用程序开始使用这个队列？如何确保一个(而且只有一个)工作人员将从这个队列中消费？

Answer 1

这样的特性不是由rabbitMQ提供的。但是，您可以通过以下方式实现它。但是，您将不得不使用轮询，而轮询效率不高(与订阅/发布相比)。你还必须利用动物园管理员来协调不同的工作人员。

您将创建两个队列:一个高优先级队列(用于管理作业)和一个低优先级队列(用于正常用户作业)。这10个工作人员将从两个队列中检索消息。每个工作人员将执行一个无限循环(理想情况下，队列为空时的睡眠间隔)，在该循环中，它将尝试可转换地从每个队列检索一条消息：

• 对于高优先级队列，工作人员只检索消息、处理消息并向队列确认。
• 对于低优先级队列，工作人员试图在中持有锁(通过写入特定的文件-znode)，如果成功，则读取消息，处理并确认消息。如果动物园管理员写入失败，其他人持有锁，因此此工作人员跳过此步骤并重复循环。

Answer 2

正如Dimos所说，你需要自己构建一些东西来实现这一点。下面是另一种实现，它需要额外的队列和一些持久存储。

• 以及现有的作业队列，创建一个“可处理作业队列”。只有满足业务规则的作业才会添加到此队列中。
• 为作业队列创建一个使用者(名为“限制器”)。限制器还需要持久存储(例如Redis或关系数据库)来记录当前正在处理的作业。限制器从作业队列中读取并写入可处理的作业队列。
• 当员工应用程序完成处理作业时，它将“作业完成”事件添加到作业队列中。 (五)

限制器的逻辑如下：

• 当收到作业消息时，请检查持久存储，以查看作业是否已为当前用户运行：。
  • 如果没有，将存储中的作业记录为正在运行，并将作业消息添加到可处理的作业队列中。
  • 如果正在运行现有作业，则将存储中的作业记录为挂起的作业。
  • 如果作业是管理用户的，请始终将其添加到可处理的作业队列中。

​

• 当接收到“作业完成”消息时，从持久存储中的“运行作业”列表中删除该作业。然后检查该用户的挂起作业的存储空间：
  • 如果找到作业，请将作业的状态从挂起更改为运行，并将其添加到可处理的作业队列中。
  • 否则什么也不做。

​

• 一次只能运行限制程序的一个实例。这可以通过仅启动限制器进程的单个实例，或者通过在持久存储中使用锁定机制来实现。

这是相当重要的，但如果您需要查看正在发生的事情，则始终可以检查持久存储。