1. 延时消息

延时消息是指消费者延时消费，比如电商里，提交了一个订单就可以发送一个延时消息，1h后去检查这个订单的状态，如果还是未付款就取消订单释放库存。

它的实现与普通消息的发送和消费没多大区别，只多了一句话：

message.setDelayTimeLevel()

1.1 使用限制

现在RocketMq并不支持任意时间的延时，需要设置几个固定的延时等级，从1s到2h分别对应着等级1到18

private String messageDelayLevel = "1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h";

1.2 示例

1.2.1 延时消息的生产者

public static void main(String[] args) throws MQClientException, RemotingException, InterruptedException, MQBrokerException {// 创建 producer 实例，并设置生产者组名DefaultMQProducer producer = new DefaultMQProducer("delayGroup");// 设置 NameServer 地址producer.setNamesrvAddr("127.0.0.1:9876");// 启动生产者producer.start();for (int i = 0; i < 3; i++) {Message message = new Message("delayTopic", "delayTag", ("Hello Delay message-" + i).getBytes());// 设置延时等级2,这个消息将在5s之后发送(现在只支持固定的几个时间,详看delayTimeLevel)message.setDelayTimeLevel(2);// 发送消息SendResult sendResult = producer.send(message);// 打印结果System.out.println(String.format("SendResult status:%s, queueId:%d",sendResult.getSendStatus(),sendResult.getMessageQueue().getQueueId()));}// 关闭生产者producer.shutdown();
}

打印结果：

SendResult status:SEND_OK, queueId:2
SendResult status:SEND_OK, queueId:3
SendResult status:SEND_OK, queueId:0

1.2.2 延时消息的消费者

public static void main(String[] args) throws MQClientException {// 实例化消费者，并设置组名DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("delayGroup");// 设置 NameServer 地址consumer.setNamesrvAddr("127.0.0.1:9876");// 订阅 Topicconsumer.subscribe("delayTopic", "delayTag");// 注册消息监听者consumer.registerMessageListener(new MessageListenerConcurrently() {@Overridepublic ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> messages, ConsumeConcurrentlyContext context) {for (MessageExt message : messages) {// 打印消息内容和延时时间System.out.println("Receive message[msgBody=" + new String(message.getBody()) + "]");}return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;}});// 启动消费者consumer.start();
}

在5秒钟之后就会收到延时消息，打印结果：

Receive message[msgBody=Hello Delay message-0]
Receive message[msgBody=Hello Delay message-1]
Receive message[msgBody=Hello Delay message-2]

2. 批量消息

批量发送消息能显著提高传递小消息的性能。限制是这些批量消息应该有相同的 topic，相同的waitStoreMsgOK，而且不能是延时消息。此外，这一批消息的总大小不应超过4MB。

2.1 小于4MB的批量消息发送

public static void main(String[] args) throws MQClientException, RemotingException, InterruptedException, MQBrokerException {// 创建 producer 实例，并设置生产者组名DefaultMQProducer producer = new DefaultMQProducer("batchGroup");// 设置 NameServer 地址producer.setNamesrvAddr("127.0.0.1:9876");// 启动生产者producer.start();// 创建批量消息List<Message> messageList = new ArrayList<>();Message message1 = new Message("batchTopic", "batchTag", ("Hello Batch message-1").getBytes());Message message2 = new Message("batchTopic", "batchTag", ("Hello Batch message-2").getBytes());Message message3 = new Message("batchTopic", "batchTag", ("Hello Batch message-3").getBytes());messageList.add(message1);messageList.add(message2);messageList.add(message3);// 发送批量消息SendResult sendResult = producer.send(messageList);// 打印结果System.out.println(String.format("SendResult status:%s, queueId:%d",sendResult.getSendStatus(),sendResult.getMessageQueue().getQueueId()));// 关闭生产者producer.shutdown();
}

2.2 大于4MB的批量消息发送

要发送大于4MB的消息，我们只需要把大的消息分裂成若干个小的消息。首先创建一个拆分消息的工具类

public class ListSplitter implements Iterator<List<Message>> {private final int SIZE_LIMIT = 1024 * 1024 * 4;private final List<Message> messages;private int currIndex;public ListSplitter(List<Message> messages) {this.messages = messages;}@Overridepublic boolean hasNext() {return currIndex < messages.size();}@Overridepublic List<Message> next() {int nextIndex = currIndex;int totalSize = 0;for (; nextIndex < messages.size(); nextIndex++) {Message message = messages.get(nextIndex);int tmpSize = message.getTopic().length() + message.getBody().length;Map<String, String> properties = message.getProperties();for (Map.Entry<String, String> entry : properties.entrySet()) {tmpSize += entry.getKey().length() + entry.getValue().length();}// 增加日志的开销20字节tmpSize = tmpSize + 20;if (tmpSize > SIZE_LIMIT) {//单个消息超过了最大的限制//忽略,否则会阻塞分裂的进程if (nextIndex - currIndex == 0) {//假如下一个子列表没有元素,则添加这个子列表然后退出循环,否则只是退出循环nextIndex++;}break;}if (tmpSize + totalSize > SIZE_LIMIT) {break;} else {totalSize += tmpSize;}}List<Message> subList = messages.subList(currIndex, nextIndex);currIndex = nextIndex;return subList;}
}

然后在实现发送消息，和上面的区别在于在发送消息前，使用了工具类分隔消息

public static void main(String[] args) throws MQClientException, RemotingException, InterruptedException, MQBrokerException {// 创建 producer 实例，并设置生产者组名DefaultMQProducer producer = new DefaultMQProducer("batchGroup");// 设置 NameServer 地址producer.setNamesrvAddr("127.0.0.1:9876");// 启动生产者producer.start();// 创建批量消息List<Message> messageList = new ArrayList<>();Message message1 = new Message("batchTopic", "batchTag", ("Hello Batch message-1").getBytes());Message message2 = new Message("batchTopic", "batchTag", ("Hello Batch message-2").getBytes());Message message3 = new Message("batchTopic", "batchTag", ("Hello Batch message-3").getBytes());messageList.add(message1);messageList.add(message2);messageList.add(message3);//发送批量消息：把大的消息分裂成若干个小的消息ListSplitter splitter = new ListSplitter(messageList);while (splitter.hasNext()) {try {List<Message> listItem = splitter.next();SendResult sendResult = producer.send(listItem);System.out.println(String.format("SendResult status:%s, queueId:%d",sendResult.getSendStatus(),sendResult.getMessageQueue().getQueueId()));} catch (Exception e) {e.printStackTrace();//处理error}}// 关闭生产者producer.shutdown();
}

2.3 消费者代码

消费者代码与普通消费者代码实现一样

public static void main(String[] args) throws MQClientException {// 实例化消费者，并设置组名DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("batchGroup");// 设置 NameServer 地址consumer.setNamesrvAddr("127.0.0.1:9876");// 订阅 Topicconsumer.subscribe("batchTopic", "batchTag");// 注册消息监听者consumer.registerMessageListener(new MessageListenerConcurrently() {@Overridepublic ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> messages, ConsumeConcurrentlyContext context) {for (MessageExt message : messages) {// 打印消息内容System.out.println("Receive message[msgBody=" + new String(message.getBody()) + "] ");}return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;}});// 启动消费者consumer.start();
}

技术无他，唯有熟尔。
知其然，也知其所以然。
踏实一些，不要着急，你想要的岁月都会给你。

消息中间件RocketMQ的延时消息和批量消息相关推荐

springboot+rocketmq（5）:实现批量消息
一.概述 1.批量发送消息: 批量发送消息能显著提高传递小消息的性能.限制是这些批量消息应该有相同的 topic,相同的 waitStoreMsgOK,而且不能是延时消息.此外,这一批消息的总大小不应 ...
RocketMQ-初体验RocketMQ(09)-广播消息、延时消息、批量消息
文章目录广播消息广播消息概述演示步骤延时消息概述使用场景延时机制实现原理示例批量消息批量消息概述示例代码广播消息广播消息概述广播消息就是向所有用户发送消息. 如果我们希 ...
跟我学RocketMQ之批量消息发送源码解析
上篇文章跟我学RocketMQ之消息发送源码解析中,我们已经对普通消息的发送流程进行了详细的解释,但是由于篇幅问题没有展开讲解批量消息的发送.本文中,我们就一起来集中分析一下批量消息的发送是怎样的 ...
RocketMQ 消息结构和消息类型
发送消息的一方称为生产者,负责生产消息,一般由业务系统负责生产消息.一个消息生产者会把业务应用系统里产生的消息发送到 broker 服务器.RocketMQ 提供多种发送方式,同步发送.异步发送.顺序 ...
消息中间件RocketMQ
消息中间件RocketMQ RocketMQ 是阿里巴巴开源的分布式消息中间件.支持事务消息.顺序消息.批量消息.延时消息.消息回溯等.它里面有几个区别于标准消息中件间的概念,如Group.Top ...
rocktmq 消息延时清空_使用Kotlin+RocketMQ实现延时消息的示例代码
一. 延时消息延时消息是指消息被发送以后,并不想让消费者立即拿到消息,而是等待指定时间后,消费者才拿到这个消息进行消费. 使用延时消息的典型场景,例如: 在电商系统中,用户下完订单30分钟内没支付, ...
消息中间件：RocketMQ 介绍（特性、术语、原理、优缺点、消息顺序、消息重复）
前些天发现了一个巨牛的人工智能学习网站,通俗易懂,风趣幽默,忍不住分享一下给大家.点击跳转到教程. 消息中间件的作用 1. 应用解耦 2. 异步处理比如用户注册场景,注册主流程完成以后,需要调用邮件 ...
消息中间件学习总结（16）——17 个方面，综合对比 Kafka、RabbitMQ、RocketMQ、ActiveMQ 四个分布式消息队列
本文将从,Kafka.RabbitMQ.ZeroMQ.RocketMQ.ActiveMQ 17 个方面综合对比作为消息队列使用时的差异. 一.资料文档 Kafka:中.有kafka作者自己写的书,网上 ...
[RocketMQ]消息中间件—RocketMQ消息消费（一）
2019独角兽企业重金招聘Python工程师标准>>> 文章摘要:在发送消息给RocketMQ后,消费者需要消费.消息的消费比发送要复杂一些,那么RocketMQ是如何来做的呢? 在 ...

消息中间件RocketMQ的延时消息和批量消息

文章目录