RocketMQ生产者调用start发送消息原理示例
RocketMQ发送消息
我们在使用RocketMQ发送消息时，一般都会使用DefaultMQProducer，类型的代码如下：
DefaultMQProducer producer = new DefaultMQProducer("producer_group");
producer.setNamesrvAddr("42.192.50.8:9876");
try {
producer.start();
producer.send(new Message("topic", "ping".getBytes(StandardCharsets.UTF_8)));
} catch (Exception e) {
e.printStackTrace();
} finally {
producer.shutdown();
}
上述代码中，在消息发送之前调用了start()方法，如果不调用start()方法，直接发送消息，那么会出现以下报错：
报错消息里面很明显地告知我们，目前这个DefaultMQProducer状态没有准备好，还不能发送消息。为了一探究竟，我们得去看看start()里面究竟做了什么操作呢？
start()里面究竟做了什么操作
我们根据源码一路走下来，可以追踪到DefaultMQProducerImpl.start(final boolean startFactory)这个方法：
public void start(final boolean startFactory) throws MQClientException {
switch (this.serviceState) {
case CREATE_JUST:
this.serviceState = ServiceState.START_FAILED;
this.checkConfig();
if (!this.defaultMQProducer.getProducerGroup().equals(MixAll.CLIENT_INNER_PRODUCER_GROUP)) {
this.defaultMQProducer.changeInstanceNameToPID();
}
// 创建MQClientInstance
this.mQClientFactory = MQClientManager.getInstance().getOrCreateMQClientInstance(this.defaultMQProducer, rpcHook);
// 注册Producer到MQClientInstance中
boolean registerOK = mQClientFactory.registerProducer(this.defaultMQProducer.getProducerGroup(), this);
if (!registerOK) {
this.serviceState = ServiceState.CREATE_JUST;
throw new MQClientException("The producer group[" + this.defaultMQProducer.getProducerGroup()
+ "] has been created before, specify another name please." + FAQUrl.suggestTodo(FAQUrl.GROUP_NAME_DUPLICATE_URL),
null);
}
this.topicPublishInfoTable.put(this.defaultMQProducer.getCreateTopicKey(), new TopicPublishInfo());
// 启动MQClientInstance实例
if (startFactory) {
mQClientFactory.start();
}
log.info("the producer [{}] start OK. sendMessageWithVIPChannel={}", this.defaultMQProducer.getProducerGroup(),
this.defaultMQProducer.isSendMessageWithVIPChannel());
this.serviceState = ServiceState.RUNNING;
break;
case RUNNING:
case START_FAILED:
case SHUTDOWN_ALREADY:
throw new MQClientException("The producer service state not OK, maybe started once, "
+ this.serviceState
+ FAQUrl.suggestTodo(FAQUrl.CLIENT_SERVICE_NOT_OK),
null);
default:
break;
}
上述代码主要做了以下几点：
1.创建MQClientInstance实例；
2.注册Producer到MQClientInstance实例中；
3.启动MQClientInstance实例；
MQClientInstance实例并不是每次都会创建的，它创建出来也会缓存的MQClientManager中，不过根据源码来看的话，每次创建Producer都会对应创建一个新的MQClientInstance实例，所以一般情况下不建议一个应用服务中重复创建Producer；
最终start()方法的关键实现逻辑还是需要进入MQClientInstance.start()中：
public void start() throws MQClientException {
synchronized (this) {
switch (this.serviceState) {
case CREATE_JUST:
this.serviceState = ServiceState.START_FAILED;
// 如果namesrv地址为null，那么就需要自己找namesrv地址
if (null == this.clientConfig.getNamesrvAddr()) {
this.mQClientAPIImpl.fetchNameServerAddr();
}
// 开启一个请求响应渠道，没猜错的话，应该是netty实现的
this.mQClientAPIImpl.start();
// 开启定时任务
this.startScheduledTask();
// 开启拉消息服务
this.pullMessageService.start();
// 开启负载均衡服务
this.rebalanceService.start();
// 再开启一个默认生产者，这个生产者不需要启动MQClientInstance实例
this.defaultMQProducer.getDefaultMQProducerImpl().start(false);
log.info("the client factory [{}] start OK", this.clientId);
this.serviceState = ServiceState.RUNNING;
break;
case START_FAILED:
throw new MQClientException("The Factory object[" + this.getClientId() + "] has been created before, and failed.", null);
default:
break;
}
}
}
看样子，这才是start()方法真正要做的事情：
1.找namesrv地址，应该是后面需要使用namesrv地址查询对应的broker；
2.开启Netty客户端的初始化，包括与namesrv建立信道；另外开启两个定时任务，一个清除列表中过期的请求，第二个就是筛选可用的namesrv服务；
3.开启一些定时任务；包括如果没有设置namesrv地址的话，会从指定站点拉namesrv地址；清除下线broker并发送心跳给所有的broker等工作；
4.因为当前是生产者，所以pullMessageService很快就结束；
5.生产者不需要做负载均衡，所以rebalanceService很快也结束；
6.给默认创建的生产者执行一下start()方法，其实啥也没做；
上述大多数任务都是给消费者使用的，作为生产者，唯一起作用的就是前三步，查找namesrv地址、第二步与namesrv建立通信以及第三步对broker的一些定时清理工作；不过没有发生消息之前，是不会从远程获取任何数据的。所以综上所述，start()方法里面只做了以下两件事情：
1.与namesrv建立通信渠道，它甚至都没有从namesrv获取任何数据；
2.启动一些定时任务，包括清理下线的broker；
小结
虽然在生产者中，start()方法里面真正做的事情比较少，但是却是非常有必要的。发送消息之前，我们没有使用start()方法，导致消息发送失败，是因为生产者与namesrv之间的通信渠道没有建立。
以上就是RocketMQ生产者调用start发送消息原理示例的详细内容，更多关于RocketMQ调用start发送消息的资料请关注脚本之家其它相关文章！
相关文章
这篇文章主要介绍了Spring中@Autowired自动注入map详解, spring是支持基于接口实现类的直接注入的,支持注入map,list等集合中,不用做其他的配置,直接注入,需要的朋友可以参考下2023-10-10
这篇文章主要为大家介绍了java为什么需要虚拟机jvm的原理详解，有需要的朋友可以借鉴参考下，希望能够有所帮助，祝大家多多进步早日升职加薪2021-11-11
本篇文章主要介绍了Spring batch 入门学习教程(附源码)，小编觉得挺不错的，现在分享给大家，也给大家做个参考。一起跟随小编过来看看吧2017-11-11
构造方法用于创建和初始化类对象，也就是说，构造方法负责”生出“一个类对象，并可以在对象出生时进行必要的操作，在这篇文章中会给大家简单介绍一下析构方法,需要的朋友可以参考下2023-05-05
Apache的Groovy是Java平台上设计的面向对象编程语言，这门动态语言拥有类似Python、Ruby和Smalltalk中的一些特性，可以作为Java平台的脚本语言使用，这篇文章主要介绍了Java 热更新 Groovy 实践及踩坑指南,需要的朋友可以参考下2022-09-09
这篇文章主要介绍了Java开发工具IntelliJ IDEA安装图解,文中通过示例代码介绍的非常详细，对大家的学习或者工作具有一定的参考学习价值,需要的朋友可以参考下2019-11-11
Intellj idea新建的java源文件夹不是蓝色的图文解决办法
idea打开java项目后新建的模块中,java文件夹需要变成蓝色,这篇文章主要给大家介绍了关于Intellj idea新建的java源文件夹不是蓝色的相关资料,文中通过图文介绍的非常详细,需要的朋友可以参考下2024-02-02
这篇文章主要为大家详细介绍了Java实现按行分割大文件，文中示例代码介绍的非常详细，具有一定的参考价值，感兴趣的小伙伴们可以参考一下2020-05-05
这篇文章主要介绍了Java序列化原理详解，文章围绕主题展开详细的内容介绍，具有一定的参考价值，感兴趣的小伙伴可以参考一下2022-06-06
深入探究 spring-boot-starter-parent的作用
这篇文章主要介绍了spring-boot-starter-parent的作用详解,本文给大家介绍的非常详细，对大家的学习或工作具有一定的参考借鉴价值，感兴趣的小伙伴可以跟着小编一起来学习一下2023-05-05
最新评论