RabbitMQ：一个工作人员听很多队列但一次收到一条消息

Question

有没有一种方法可以让我设置一个监听多个队列但一次只处理来自单个队列的单个消息的工作人员？

用例： 1. 任务A 2. 任务 B 3.任务C

我想要 10 个工人。 #1 - #3 可以处理Task A，#1 - #6 可以处理Task B，它们都可以处理Task C。

我曾经让我的所有工作人员监听所有队列，但是当一些任务无法被太多工作人员处理时（例如，向另一个服务发出太多并发请求），它们会导致问题。

让工作人员在队列为空而其他队列已满时监听单个队列会浪费其资源。

Answer 1

根据要求，工作人员可以一次处理任何一项任务，我建议为所有 3 种类型的消息创建一个队列，并将所有 10 个工作人员作为这个公共队列的接收者。 您可以使用自定义标头AMQP.BasicProperties --> Map<String,Object> headers 来区分消息。

例如，

发送者和队列：

(1) 如果您可以控制 Sender 和 Queues：

制作单个队列而不是 3 个不同的队列，如下所示。 发件人将在发布之前为每条消息添加自定义标头。 headers.put("type","Task#")

(2) 如果您无法控制 Sender 和 Queues：

您可以使用Processor 将headers.put("type","Task#") 添加到每条消息中，并将其发送至common queue。如下图。

工人：

每个工作人员都可以检查接收到的消息的类型并做出相应的行为。 Worker 4,5,6 只能处理 TaskB 和 TaskC。一次将处理任何人的消息。 （如下sn-p中提到的——仅供参考）

//Worker 4,5,6  (sample steps)
public void handler(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) {
    if(properties.getHeaders().get("type").equals("TaskA")) {
        // Negative acknowledgement - the message will be requeue.
    }
    else if(properties.getHeaders().get("type").equals("TaskB")) {
        // Convert body into TaskB (if required)
        // Process Task B
        // Positive acknowledgement
    }
    else if(properties.getHeaders().get("type").equals("TaskC")) {
        // Convert body into TaskC (if required)
        // Process Task C
        // Positive acknowledgement
    }
}