一、发送 & 接收批量消息
1.1、概述
批量消息是指RocketMQ可以把一组消息集合一次性发送,这一组消息会被当做一个消息供消费者消费。
1.2、Demo05MQTestApp
/**
* @Author : 一叶浮萍归大海
* @Date: 2023/12/25 11:48
* @Description: 发送 & 接收批量消息
*/
@Slf4j
public class Demo05MQTestApp {
/**
* 发送批量消息
*/
@Test
public void demo5Producer() throws Exception {
// 1、创建一个生产者
DefaultMQProducer producer = new DefaultMQProducer("batch-producer-group");
// 2、连接NameServer
producer.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方式)
*/
@Test
public void demo5PushConsumer() throws Exception {
// 1、创建一个消费者
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("batch-consumer-group");
// 2、连接NameServer
consumer.setNamesrvAddr(RocketMQConstant.NAME_SERVER_ADDR);
// 3、订阅消息,*表示订阅该主题所有的消息
consumer.subscribe("batch-topic", "*");
// 4、设置监听器(采用异步回调方式,一直监听)
consumer.registerMessageListener(new MessageListenerConcurrently() {
@Override
public 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、挂起当前JVM
System.in.read();
}
}
1.3、测试
先后运行demo5PushConsumer、demo5Producer,观察控制台日志输出信息。