15

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

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

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

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

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

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

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

4

2 回答 2

6

正如 Dimos 所说，您需要自己构建一些东西来实现这一点。这是一个替代实现，它需要一个额外的队列和一些持久存储。

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

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

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

限制器的逻辑如下：

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

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

于 2016-10-13T06:32:43.740 回答
4

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

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

  • 对于高优先级队列，worker 只是检索一条消息，处理它并向队列确认。
  • 对于低优先级队列，worker 尝试在 Zookeeper 中持有锁（通过写入特定的文件-znode），如果成功，则读取消息，处理并确认。如果zookeeper写不成功，其他人持有锁，所以这个worker跳过这一步，重复循环。
于 2016-10-12T20:23:04.570 回答