RocketMQ使用批量消息

Posted 2022-01-07 乐观男孩

目录

• 说明
• 生产端
• 消费端
• 遗留问题

说明

批量发送消息条件：
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]，其中的原因暂未深入研究，后期有时间再看吧，如果哪位兄弟知道原因也请留言哦~