一、发送 & 接收单向消息
1.1、概述
发送单向消息,适用于发送方不关心或者不在意消息的发送结果,这种方式的吞吐量很大,但是存在消息丢失的风险,对于重要消息要慎用!该种方式通常适用于对消息没有那么严格的场景中,例如日志信息的采集。
1.2、Demo03MQTestApp
/*** @Author : 一叶浮萍归大海* @Date: 2023/12/25 10:12* @Description: 发送 & 接收单向消息*/
@Slf4j
public class Demo03MQTestApp {/*** 发送同步消息*/@Testpublic void demo3Producer() throws Exception {// 1、创建一个生产者DefaultMQProducer producer = new DefaultMQProducer("oneway-producer-group");// 2、连接NameServerproducer.setNamesrvAddr(RocketMQConstant.NAME_SERVER_ADDR);// 3、启动producer.start();// 4、创建消息Message message = new Message("oneway-topic", "这是一个单向消息)".getBytes(StandardCharsets.UTF_8));// 5、发送消息producer.sendOneway(message);log.info("【demo3Producer】发送消息成功!");// 6、关闭生产者producer.shutdown();}/*** 接收单向消息(Push方式)*/@Testpublic void demo3PushConsumer() throws Exception {// 1、创建一个消费者DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("oneway-consumer-group");// 2、连接NameServerconsumer.setNamesrvAddr(RocketMQConstant.NAME_SERVER_ADDR);// 3、订阅消息,*表示订阅该主题所有的消息consumer.subscribe("oneway-topic", "*");// 4、设置监听器(采用异步回调方式,一直监听)consumer.registerMessageListener(new MessageListenerConcurrently() {@Overridepublic ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {// 业务逻辑log.info("我是消费者【demo3PushConsumer】");for (MessageExt msg : msgs) {log.info("我是消费者【demo3PushConsumer】,我收到的消息是:{},当前时间:{}", 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("【demo3PushConsumer】启动成功,正在等待接收消息...");// 6、挂起当前JVMSystem.in.read();}}
1.3、测试
先后运行demo3PushConsumer和demo3Producer,观察控制台日志: