在上篇介绍了如何简单的发送一个消息队列之后，我们本篇来看下RabbitMQ的另外一种模式，工作队列。
什么是工作队列
我们上篇文章说的是，一个生产者生产了消息被一个消费者消费了，如下图
上面这种简单的消息队列确实可以处理我们的任务，但是当我们队列中的任务过多，处理每条任务有需要很长的耗时，那么使用一个消费者处理消息显然不不够的，所以我们可以增加消费者，来共享消息队列中的消息，进行任务处理。
也就是如下图
虽然上图我只花了一个生产者A，那么同理，能有多个消费者，那也能多个生产者。
代码
发送消息
代码语言：javascript复制public class Send {
public static final String QUEUE_NAME = "test_word_queue";
public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {
// 获取连接
Connection connection = MQConnectUtil.getConnection();
// 创建通道
Channel channel = connection.createChannel();
// 声明队列
channel.queueDeclare(QUEUE_NAME, false, false, false, null);
// 模拟发送20条消息
for (int i = 0; i < 20; i ) {
String msg = "消息：" i;
channel.basicPublish("", QUEUE_NAME, null, msg.getBytes());
Thread.sleep(i * 20);
System.out.println(msg);
}
channel.close();
connection.close();
}
}
消费者A
代码语言：javascript复制public class Consumer1 {
public static final String QUEUE_NAME = "test_word_queue";
public static void main(String[] args) throws Exception {
// 获取连接
Connection connection = MQConnectUtil.getConnection();
// 创建频道
Channel channel = connection.createChannel();
// 队列声明
channel.queueDeclare(QUEUE_NAME, false, false, false, null);
// 定义消费者
DefaultConsumer consumer = new DefaultConsumer(channel) {
@SneakyThrows
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) {
String msg = new String(body, StandardCharsets.UTF_8);
System.out.println("消费者[A]-内容：" msg);
Thread.sleep(2 * 1000);
}
};
// 监听队列
channel.basicConsume(QUEUE_NAME, true, consumer);
}
}
消费者B
代码语言：javascript复制public class Consumer2 {
public static final String QUEUE_NAME = "test_word_queue";
public static void main(String[] args) throws Exception {
// 获取连接
Connection connection = MQConnectUtil.getConnection();
// 创建频道
Channel channel = connection.createChannel();
// 队列声明
channel.queueDeclare(QUEUE_NAME, false, false, false, null);
// 定义消费者
DefaultConsumer consumer = new DefaultConsumer(channel) {
@SneakyThrows
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) {
String msg = new String(body, StandardCharsets.UTF_8);
System.out.println("消费者[B]-内容：" msg);
Thread.sleep(1000);
}
};
// 监听队列
channel.basicConsume(QUEUE_NAME, true, consumer);
}
}
我们来看下消费者A和消费者B的消费情况
- 消费者B
- 消费者A
有没有发现什么问题，我总共模拟发送了20条消息，细心的同学可以发现，消费者A和消费者B消费了同样多的消息，都消费了10天，但是我在消费者A和消费者B中,什么sleep不通的时长,按道理说消费者B要比消费者A处理消息的速度快，处理的消息更多，那么为什么会产生这样的原因？
RabbitMQ工作队列的默认配置
默认情况下，RabbitMQ会将每个消息依次发送给下一个消费者，每个消费者收到的消息数量其实是一样的，我们把这种分发消息的方式称为轮训分发模式。
本篇我们就简单介绍这么多内容，有心学习的童鞋一定要敲敲代码，看不一定能看会，只有自己敲一遍，才能有所理解。
日拱一卒,功不唐捐
