当前位置：首页 > news >正文

系列十（实战）、发送接收批量消息（Java操作RocketMQ）

news 来源：原创 2024/5/9 13:42:06

一、发送 & 接收批量消息

1.1、概述

批量消息是指RocketMQ可以把一组消息集合一次性发送，这一组消息会被当做一个消息供消费者消费。

1.2、Demo05MQTestApp

/*** @Author : 一叶浮萍归大海* @Date: 2023/12/25 11:48* @Description: 发送 & 接收批量消息*/
@Slf4j
public class Demo05MQTestApp {/*** 发送批量消息*/@Testpublic void demo5Producer() throws Exception {// 1、创建一个生产者DefaultMQProducer producer = new DefaultMQProducer("batch-producer-group");// 2、连接NameServerproducer.setNamesrvAddr(RocketMQConstant.NAME_SERVER_ADDR);// 3、启动producer.start();// 4、创建批量消息List<Message> messages = Arrays.asList(new Message("batch-topic", "我是一组消息中的A消息".getBytes(StandardCharsets.UTF_8)),new Message("batch-topic", "我是一组消息中的B消息".getBytes(StandardCharsets.UTF_8)),new Message("batch-topic", "我是一组消息中的C消息".getBytes(StandardCharsets.UTF_8)));// 5、发送消息producer.send(messages);log.info("【demo5Producer】发送消息成功!");// 6、关闭生产者producer.shutdown();}/*** 接收批量消息（Push方式）*/@Testpublic void demo5PushConsumer() throws Exception {// 1、创建一个消费者DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("batch-consumer-group");// 2、连接NameServerconsumer.setNamesrvAddr(RocketMQConstant.NAME_SERVER_ADDR);// 3、订阅消息，*表示订阅该主题所有的消息consumer.subscribe("batch-topic", "*");// 4、设置监听器（采用异步回调方式，一直监听）consumer.registerMessageListener(new MessageListenerConcurrently() {@Overridepublic ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {// 业务逻辑log.info("我是消费者【demo5PushConsumer】");for (MessageExt msg : msgs) {log.info("我是消费者【demo5PushConsumer】，我收到的消息是：{}，当前时间：{}", StrUtil.utf8Str(msg.getBody()),LocalDateTimeUtil.format(LocalDateTime.now(),"yyyy-MM-dd HH:mm:ss"));}/*** 返回值：消费消息成功与否*      CONSUME_SUCCESS：表明消费成功，消息会从MQ出队*      RECONSUME_LATER：表明消费失败，消息会重新回到队里，过一会儿再重新投递出来给当前消费者或者其他消费者*/return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;}});// 5、启动consumer.start();log.info("【demo5PushConsumer】启动成功，正在等待接收消息...");// 6、挂起当前JVMSystem.in.read();}}