企业网站在哪里建,软文写作要求,深圳人才网58同城招聘,wordpress升级快速一、发送 接收单向消息
1.1、概述 发送单向消息#xff0c;适用于发送方不关心或者不在意消息的发送结果#xff0c;这种方式的吞吐量很大#xff0c;但是存在消息丢失的风险#xff0c;对于重要消息要慎用#xff01;该种方式通常适用于对消息没有那么严格的场景中…一、发送 接收单向消息
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(ListMessageExt 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观察控制台日志
