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

场景（我已经简化了事情）：

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

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

问题：如果用户在同一分钟内启动 10 个作业，并且一天中的那个时间只有 10 个工作应用程序启动，则该最终用户实际上接管了自己的所有计算时间。

问题：如何确保在任何时候每个最终用户只处理一项作业？ （Bonus：某些最终用户（例如管理员）不得受到限制）

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

解决方案？：我应该为每个最终用户动态创建一个自动删除独占队列吗？如果是，我如何告诉工作应用程序开始使用这个队列？如何确保一名（且仅一名）工作人员将从该队列中消费？

正如迪莫斯所说，你需要自己构建一些东西来实现这一点。这是一种替代实现，需要额外的队列和一些持久存储。

• 除了现有的作业队列之外，还创建一个“可处理作业队列”。只有满足您的业务规则的作业才会添加到此队列中。
• 为作业队列创建一个使用者（名为“Limiter”）。 Limiter还需要持久存储（例如Redis或关系数据库）来记录当前正在处理哪些作业。限制器从作业队列读取并写入可处理作业队列。

• 当工作应用程序完成处理作业时，它会将“作业完成”事件添加到作业队列中。

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

限制器的逻辑如下：

• When a job message is received, check the persistent storage to see if a job is already running for the current user:
  • 如果没有，则将存储中的作业记录为正在运行，并将作业消息添加到可处理作业队列中。
  • 如果现有作业正在运行，则将该作业在存储中记录为待处理作业。
  • 如果作业是针对管理员用户的，请始终将其添加到可处理作业队列中。
• When a "job finished" message is received, remove that job from the "running jobs" list in the persistent storage. Then check the storage for a pending job for that user:
  • 如果找到作业，则将该作业的状态从待处理更改为正在运行，并将其添加到可处理作业队列中。
  • 否则，什么也不做。
• 限制器进程一次只能运行一个实例。这可以通过仅启动限制器进程的单个实例或通过使用持久存储中的锁定机制来实现。

它相当重量级，但如果您需要查看发生了什么，您可以随时检查持久存储。