说明

批量发送消息条件：
1、发送到同一个topic
2、等待同一个发送结果
3、不允许使用定时消息
4、不支持半事务特性
5、同一批消息大小不能超过1MB，否则需要自己进行切割

发送批量消息，最主要的区别是在发送消息的send方法入参一个List。

生产端

@Test
public void sendMessage() throws Exception {
DefaultMQProducer defaultMQProducer = RocketMqUtil.getDefaultMQProducer();
Message message1 = new Message(RocketMqUtil.TOPIC, "batch-1", "batch-message-1".getBytes(Charset.forName("UTF-8")));
Message message2 = new Message(RocketMqUtil.TOPIC, "batch-2", "batch-message-2".getBytes(Charset.forName("UTF-8")));
Message message3 = new Message(RocketMqUtil.TOPIC, "batch-3", "batch-message-3".getBytes(Charset.forName("UTF-8")));
//批量发送消息
List<Message> messages = Lists.newArrayList(message1, message2, message3);
SendResult sendResult = defaultMQProducer.send(messages);
log.info("发送消息结果：{}", sendResult.getSendStatus().name());
defaultMQProducer.shutdown();
}

消费端

@Test
public void consumer() throws Exception {
DefaultMQPushConsumer defaultMQPushConsumer = RocketMqUtil.getDefaultMQPushConsumer();
defaultMQPushConsumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
defaultMQPushConsumer.subscribe(RocketMqUtil.TOPIC, "*");
defaultMQPushConsumer.registerMessageListener(new MessageListenerConcurrently() {
@Override
public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
msgs.stream().map(MessageExt::getBody).map(String::new).forEach(System.out::println);
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
}
});
defaultMQPushConsumer.start();
Thread.sleep(5000L);
defaultMQPushConsumer.shutdown();
}

遗留问题

本地测试发送批量消息后，在rocketmq-console上看不到消息的Tag，在命令行看到的Tag为：[tagCLUSTERDefaultCluster]，其中的原因暂未深入研究，后期有时间再看吧，如果哪位兄弟知道原因也请留言哦~

最后

以上就是机灵未来为你收集整理的RocketMQ使用批量消息说明生产端消费端遗留问题的全部内容，希望文章能够帮你解决RocketMQ使用批量消息说明生产端消费端遗留问题所遇到的程序开发问题。

如果觉得靠谱客网站的内容还不错，欢迎将靠谱客网站推荐给程序员好友。