rabbitMQ 从 2 个队列消费

Question

我必须编写重负载系统，而且任务很简单。所以我决定把这个任务分成不同位置（或云）的多个工人。为了交流我想使用rabbitmq队列。

在我的系统中会有两种软件节点：调度器和工作器。调度程序将从 queue_input 获取用户输入，将其拆分为较小的任务并将这个较小的任务放入 workers_queue。工人读取这个队列并“做事”。我在这里使用了循环负载平衡 - 只要有一些工作人员崩溃，所有工作都很好。然后我会丢失有关任务完成的信息（不允许两次执行单个操作，每个任务包含一组 50 次迭代，用不同的数据执行工作代码）。

我考虑了类似 Technical_queue 之类的东西 - 调度员与工作人员通信的另一个渠道，我想知道如何以一种好的方式设计它。我使用了rabbitmq页面的教程，所以我的工作线程看起来像：

while(true) {
   message = consume(QUEUE,...);
   handle(message); //do 50 simple tasks in loop for data in message
}

如何处理第二个队列？另一个线程我们一些 while(true) {} 循环？还是有更好的解决方案？也许我应该通过主题交换重用现有队列？ （但我希望有独立的沟通方式，同时处理任务，这可能需要一些时间。

Answer 1

您可能应该看看spring-amqp (doc)。我不想告诉你添加一个层，但是 spring 库通过其SimpleMessageListenerContainer 处理线程问题和线程管理。每个容器进入一个队列，您可以指定每个队列的线程数（即工作线程）。

或者，您可以使用ExecutorService 自己制作，但您最终可能会重写SimpleMessageListenerContainer 所做的事情。此外，您还可以（通过操作系统或批处理脚本）执行更多进程，这将为每个队列添加更多消费者。

就队列拓扑而言，它完全取决于业务逻辑/关注点，通常较少取决于性能需求。更多时候，出于业务原因，您有更多队列，出于性能原因有更多工作人员，但如果队列使用相同类型的消息进行备份，请考虑为该类型的消息提供自己的队列。您的描述听起来像是两个队列，您的 worker 队列上有多个消费者。

除了线程问题和队列拓扑，我不完全确定您还问什么。

Answer 2

我建议您创建第二个队列消费者

consumer1 -> queue_process
consumer2 -> queue_process

两个消费者都应该收听同一个队列。 问候，希望对您有所帮助