RocketMQ发送同步异步单向消息
Posted 乐观男孩
tags:
篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了RocketMQ发送同步异步单向消息相关的知识,希望对你有一定的参考价值。
目录
说明
后续打算根据RocketMQ官网的例子,也对各种类型的消息进行简单的使用。在熟悉基本使用,了解各类消息的特点之后,再抽时间进行内部原理剖析。后续所有的文章,都是基于RocketMQ-4.9.1 release版本。
基础工具类
在接下来的例子中,都会用到以下工具类RocketMqUtil,其主要作用是封装了RocketMQ的NameServer地址、Topic、生产者或消费者。所以这里先贴上该工具类的代码:
package com.lzj.rocketmq;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
public class RocketMqUtil
public static final String NAME_SERVER_ADDR = "192.168.6.130:9876";
public static final String TOPIC = "topic_simple_message";
public static DefaultMQProducer getDefaultMQProducer()
try
DefaultMQProducer defaultMQProducer = new DefaultMQProducer();
defaultMQProducer.setProducerGroup("TEST_PRODUCER");
defaultMQProducer.setNamesrvAddr(NAME_SERVER_ADDR);
//同步发送消息时发送失败后的重试次数
defaultMQProducer.setRetryTimesWhenSendFailed(0);
//异步发送消息时发送失败后的重试次数
defaultMQProducer.setRetryTimesWhenSendAsyncFailed(0);
defaultMQProducer.start();
return defaultMQProducer;
catch (Exception e)
e.printStackTrace();
throw new RuntimeException("初始化RocketMQ生产者时发生异常");
public static DefaultMQPushConsumer getDefaultMQPushConsumer() throws MQClientException
try
DefaultMQPushConsumer defaultMQPushConsumer = new DefaultMQPushConsumer();
defaultMQPushConsumer.setConsumerGroup("TEST_CONSUMER");
defaultMQPushConsumer.setNamesrvAddr(NAME_SERVER_ADDR);
return defaultMQPushConsumer;
catch (Exception e)
e.printStackTrace();
throw new RuntimeException("初始化RocketMQ消费者时发生异常");
测试代码
由于代码比较简单,所以直接了当,上源码:
package com.lzj.rocketmq;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.remoting.exception.RemotingException;
import org.junit.jupiter.api.Test;
import org.springframework.boot.test.context.SpringBootTest;
import java.nio.charset.Charset;
import java.util.List;
@SpringBootTest
@Slf4j
public class SimpleMessageTest
/**
* 发送同步消息
* @throws InterruptedException
* @throws RemotingException
* @throws MQClientException
* @throws MQBrokerException
*/
@Test
public void sendMessageSynchronously() throws InterruptedException, RemotingException, MQClientException, MQBrokerException
DefaultMQProducer defaultMQProducer = RocketMqUtil.getDefaultMQProducer();
Message message = new Message(RocketMqUtil.TOPIC, "send_message_synchronously", "同步消息".getBytes(Charset.forName("UTF-8")));
SendResult sendResult = defaultMQProducer.send(message);
log.info("同步消息,结果名称:,结果值:", sendResult.getSendStatus().name(),
sendResult.getSendStatus().ordinal());
defaultMQProducer.shutdown();
/**
* 发送异步消息
* @throws RemotingException
* @throws MQClientException
* @throws InterruptedException
*/
@Test
public void sendMessageAsynchronously() throws RemotingException, MQClientException, InterruptedException
DefaultMQProducer defaultMQProducer = RocketMqUtil.getDefaultMQProducer();
Message message = new Message(RocketMqUtil.TOPIC, "send_message_asynchronously", "异步消息".getBytes(Charset.forName("UTF-8")));
defaultMQProducer.send(message, new SendCallback()
@Override
public void onSuccess(SendResult sendResult)
log.info("send message success." + sendResult.getSendStatus().ordinal());
@Override
public void onException(Throwable throwable)
log.info("send message fail.");
);
Thread.sleep(5000L);
defaultMQProducer.shutdown();
/**
* 单向传输,客户端只管发送,不等待响应,存在一定的不安全性,一般使用于记录日志的功能
* @throws RemotingException
* @throws MQClientException
* @throws InterruptedException
*/
@Test
public void sendMessageOneway() throws RemotingException, MQClientException, InterruptedException
DefaultMQProducer defaultMQProducer = RocketMqUtil.getDefaultMQProducer();
Message message = new Message(RocketMqUtil.TOPIC, "send_message_oneway", "单向消息".getBytes(Charset.forName("UTF-8")));
defaultMQProducer.sendOneway(message);
defaultMQProducer.shutdown();
@Test
public void consumeMessage() throws Exception
DefaultMQPushConsumer defaultMQPushConsumer = RocketMqUtil.getDefaultMQPushConsumer();
defaultMQPushConsumer.subscribe(RocketMqUtil.TOPIC, "*");
defaultMQPushConsumer.registerMessageListener(new MessageListenerConcurrently()
@Override
public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list,
ConsumeConcurrentlyContext consumeConcurrentlyContext)
log.info("消费到消息条数:", list.size());
list.stream().map(messageExt -> new String(messageExt.getBody(), Charset.forName("UTF-8")))
.map(String::new).forEach(System.out::println);
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
);
defaultMQPushConsumer.start();
Thread.sleep(5000L);
这里展示了三类基础消息的发送及消费的例子:
**同步消息:**客户端将消息发送到Master Broker后,需要等Master Broker将消息同步到对应的Slave Broker后,Broker才会响应客户端。这种方式响应客户端比较慢,但具有较强的数据安全性,数据丢失风险小。
**异步消息:**客户端将消息发送到Master Broker后,只要Master Broker将消息处理完,Broker就响应客户端,无需等待Master Broker将消息同步到对应的Slave Broker。这种方式对客户端响应快,但存在一定的风险性:在Master Broker处理完消息响应客户端后就宕机,此时消息可能会发生丢失。
**单向消息:**客户端只负责消息发送,不等待Broker回应,这种方式效率最高,但消息丢失的风险也最高,一般用在可以容忍消息有一定丢失的场景。
效果展示
从RocketMQ控制台上看到消息确实发送成功(测试时同步消息发了两次)
源码消费者也同样的成功消费到数据:
以上是关于RocketMQ发送同步异步单向消息的主要内容,如果未能解决你的问题,请参考以下文章
RocketMQ源码—Producer发送消息源码—单向同步异步发送消息一万字
RocketMQ源码—Producer发送单向同步异步消息源码一万字