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获得消息很少。
