如何根据条件限制并发消息使用

Question

场景(我简化了一些事情):

• 许多最终用户可以从前端Web应用程序(生产者)开始工作(繁重的工作,例如渲染大型PDF).
• 作业将发送到单个持久的RabbitMQ队列.
• 许多工作者应用程序(使用者)处理这些作业并将结果写回数据存储区.

这个相当标准的模式工作正常.

问题是:如果用户在同一分钟内启动了10个作业,并且在一天中的那个时间只有10个工作者应用程序启动,则该最终用户有效地接管了他自己的所有计算时间.

问题:如何确保每个最终用户只能处理一个作业？(奖励:一些最终用户(例如管理员)不得受到限制)

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

解决方案？:我应该为每个最终用户动态创建一个自动删除独占队列吗？如果是,我如何告诉worker应用程序开始使用此队列？如何确保一个(并且只有一个)工作者将从此队列中消耗？

Answer 1

Dimos说,你需要自己构建一些东西来实现它.这是一个替代实现,需要额外的队列和一些持久存储.

• 与现有的作业队列一样,创建"可处理的作业队列".只有满足业务规则的作业才会添加到此队列中.
• 为作业队列创建一个使用者(名为"Limiter").限制器还需要持久存储(例如Redis或关系数据库)来记录当前正在处理的作业.限制器从作业队列中读取并写入可处理的作业队列.

• 当工作器应用程序完成处理作业时,它会向作业队列添加"作业完成"事件.

  ------------     ------------     ----------- 
  | Producer | -> () job queue ) -> | Limiter | 
  ------------     ------------     ----------- 
                       ^                |                    
                       |                V                    
                       |     ------------------------       
                       |    () processable job queue )  
         job finished  |     ------------------------       
                       |                |
                       |                V
                       |     ------------------------
                       \-----| Job Processors (x10) |
                             ------------------------
  

限制器的逻辑如下:

• 收到作业消息后,检查持久存储以查看当前用户是否已在运行作业:
  • 如果不是,请将存储中的作业记录为正在运行,并将作业消息添加到可处理作业队列中.
  • 如果现有作业正在运行,请将存储中的作业记录为待处理作业.
  • 如果作业是管理员用户,请始终将其添加到可处理作业队列中.
• 收到"作业完成"消息后,从持久存储中的"正在运行的作业"列表中删除该作业.然后检查存储以查找该用户的待处理作业:
  • 如果找到作业,请将该作业的状态从挂起更改为正在运行,并将其添加到可处理作业队列中.
  • 否则,什么也不做.
• 一次只能运行一个限制器进程实例.这可以通过仅启动限制器进程的单个实例,或通过在持久存储中使用锁定机制来实现.

它是相当重量级的,但如果您需要查看正在发生的事情,您可以随时检查持久存储.

Answer 2

rabbitMQ 本身并不提供这样的功能。但是，您可以通过以下方式实现它。不过，您将不得不使用轮询，这不太有效（与订阅/发布相比）。您还必须利用 Zookeeper 来协调不同工作人员之间的关系。

您将创建 2 个队列：1 个高优先级队列（用于管理作业）和 1 个低优先级队列（用于普通用户作业）。这 10 个工作线程将从两个队列中检索消息。每个工作线程将执行一个无限循环（理想情况下，当队列为空时，有睡眠间隔），它将尝试从每个队列中交替检索消息：

• 对于高优先级队列，工作人员仅检索消息、处理它并向队列确认。
• 对于低优先级队列，工作线程尝试在 Zookeeper 中持有锁（通过写入特定的 file-znode），如果成功，则读取消息、处理该消息并确认。如果zookeeper写入不成功，则其他人持有锁，因此该worker会跳过此步骤并重复循环。