RabbitMQ如何高效的消费消息

原创
2020/05/12 09:20
阅读数 149

在上篇介绍了如何简单的发送一个消息队列之后,我们本篇来看下RabbitMQ的另外一种模式,工作队列

什么是工作队列

我们上篇文章说的是,一个生产者生产了消息被一个消费者消费了,如下图

上面这种简单的消息队列确实可以处理我们的任务,但是当我们队列中的任务过多,处理每条任务有需要很长的耗时,那么使用一个消费者处理消息显然不不够的,所以我们可以增加消费者,来共享消息队列中的消息,进行任务处理。

也就是如下图

虽然上图我只花了一个生产者A,那么同理,能有多个消费者,那也能多个生产者。

代码

发送消息

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

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

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会将每个消息依次发送给下一个消费者,每个消费者收到的消息数量其实是一样的,我们把这种分发消息的方式称为轮训分发模式

本篇我们就简单介绍这么多内容,有心学习的童鞋一定要敲敲代码,看不一定能看会,只有自己敲一遍,才能有所理解。

日拱一卒,功不唐捐




如何利用RabbitMQ 生产一个简单的消息


壹伴,简单好用的公众号编辑器插件。


壹伴,简单好用的公众号编辑器插件。


壹伴,简单好用的公众号编辑器插件。






好文章,我在看

本文分享自微信公众号 - 一个程序员的成长(xiaozaibuluo)。
如有侵权,请联系 support@oschina.cn 删除。
本文参与“OSC源创计划”,欢迎正在阅读的你也加入,一起分享。

展开阅读全文
加载中

作者的其它热门文章

打赏
0
0 收藏
分享
打赏
0 评论
0 收藏
0
分享
返回顶部
顶部