AMQP-RabbitMQ/2/工作隊列

2. 工作隊列 Work queues

Distributing tasks among workers

消息将發送給c1或者c2

個人了解

• 生産者定義Queue，并向該隊列發送消息
• 多個消費者可以從指定的同一個Queue中讀取消息。每條消息隻會發送給其中某一個消費者。
• 生産者

package com.futao.springmvcdemo.mq.rabbit.workqueue;

import com.futao.springmvcdemo.mq.rabbit.RabbitMqConnectionTools;
import com.futao.springmvcdemo.mq.rabbit.RabbitMqQueueEnum;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import lombok.Cleanup;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;

/**
 * 簡單發送者
 *
 * @author futao
 * Created on 2019-04-22.
 */
@Slf4j
public class Send {
    @SneakyThrows
    public static void main(String[] args) {
        @Cleanup
        Connection connection = RabbitMqConnectionTools.getConnection();
        @Cleanup
        Channel channel = connection.createChannel();
        //開啟持久化消息
        boolean durable = true;
        //定義一個隊列
        channel.queueDeclare(RabbitMqQueueEnum.WORK_QUEUE.getQueueName(), durable, false, false, null);
        String msg = "Hello RabbitMq!";
        for (int i = 0; i < 20; i++) {
            channel.basicPublish("", RabbitMqQueueEnum.WORK_QUEUE.getQueueName(), null, (msg + i).getBytes());
            log.info("Send msg:[{}] success", (msg + i));
        }
    }
}           

• 消費者1 - 每1秒處理一條

package com.futao.springmvcdemo.mq.rabbit.workqueue;

import com.futao.springmvcdemo.mq.rabbit.RabbitMqConnectionTools;
import com.futao.springmvcdemo.mq.rabbit.RabbitMqQueueEnum;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DeliverCallback;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;

/**
 * 簡單消費者
 *
 * @author futao
 * Created on 2019-04-22.
 */
@Slf4j
public class RecvOne {
    @SneakyThrows
    public static void main(String[] args) {
        Channel channel = RabbitMqConnectionTools.getChannel();
        //開啟持久化消息
        boolean durable = true;
        channel.queueDeclare(RabbitMqQueueEnum.WORK_QUEUE.getQueueName(), durable, false, false, null);
        log.info("Waiting for message...");
        DeliverCallback deliverCallback = ((consumerTag, message) -> {
            log.info("收到消息:[{}],tag:[{}]", new String(message.getBody()), consumerTag);
            //acknowledgment應答
            channel.basicAck(message.getEnvelope().getDeliveryTag(), false);
            try {
                Thread.sleep(1000);
            } catch (Exception e) {

            }
        });
        //關閉自動應答
        boolean autoAck = false;
        channel.basicConsume(RabbitMqQueueEnum.WORK_QUEUE.getQueueName(), autoAck, deliverCallback, consumerTag -> {
        });
    }
}           

• 消費者2 - 每2秒處理一條

package com.futao.springmvcdemo.mq.rabbit.workqueue;

import com.futao.springmvcdemo.mq.rabbit.RabbitMqConnectionTools;
import com.futao.springmvcdemo.mq.rabbit.RabbitMqQueueEnum;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DeliverCallback;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;

/**
 * 簡單消費者
 *
 * @author futao
 * Created on 2019-04-22.
 */
@Slf4j
public class RecvTwo {
    @SneakyThrows
    public static void main(String[] args) {
        Channel channel = RabbitMqConnectionTools.getChannel();
        //開啟持久化消息
        boolean durable = true;
        channel.queueDeclare(RabbitMqQueueEnum.WORK_QUEUE.getQueueName(), durable, false, false, null);
        log.info("Waiting for message...");
        DeliverCallback deliverCallback = ((consumerTag, message) -> {
            log.info("收到消息:[{}],tag:[{}]", new String(message.getBody()), consumerTag);
            //acknowledgment應答
            channel.basicAck(message.getEnvelope().getDeliveryTag(), false);
            try {
                Thread.sleep(2000);
            } catch (Exception e) {

            }
        });
        //關閉自動應答
        boolean autoAck = false;
        channel.basicConsume(RabbitMqQueueEnum.WORK_QUEUE.getQueueName(), autoAck, deliverCallback, consumerTag -> {
        });
    }
}           

• 結果
  • 消費者1日志
  • 消費者2日志
• 特點: 多個消費者之間，不論消息的處理速度，都是平均分發(公平分發)。你一個，我一個，他一個。此時是公平隊列
• 注意：
  • 定義隊列的時候，設定是否開啟消息的持久化（該設定需要同時在生産者和消費者設定）

//開啟持久化消息
        boolean durable = true;
        channel.queueDeclare(RabbitMqQueueEnum.WORK_QUEUE.getQueueName(), durable, false, false, null);           

• 如果消息隊列已經存在，則不可以修改相應的配置，必須删除原有的隊列，或者建立一個新的隊列。
• 關閉自動應答（開啟手動應答），可以防止消息在未被正确消費的情況下被Rabbitmq從隊列記憶體中删除。

實作工作隊列下的非公平隊列

消費者設定一次隻發送一條消息，并且在被正确消費之前發繼續發送下一條消息。進而使得消費快的消費者比消費慢的消費者消費更多的消息

//告訴rabbitmq一次隻發送一條消息，并且在前一個消息未被處理或者消費之前，不繼續發送下一個消息
        channel.basicQos(1);           

• 測試結果

此時明顯打破了消息的公平分發，消費快的消費者接收到的消息更多。

如果有兩個消費者，其中一個ConsumerA設定了

Qos=1

，另一個ConsumerB沒有設定。經過我的測試，ConsumerA會獲得大量的消息，都積壓在ConsumerA，而ConsumerB獲得消息很少。