Update rocketmq-producer-start.md

pull/109/head
Yang Libin 3 years ago committed by GitHub
parent 521b97b3ac
commit c4a2083959
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23

@ -1,8 +1,10 @@
# RocketMQ 生产者启动流程
入口: 入口:
org.apache.rocketmq.client.producer.DefaultMQProducer#start org.apache.rocketmq.client.producer.DefaultMQProducer#start
``` ```java
@Override @Override
public void start() throws MQClientException { public void start() throws MQClientException {
this.setProducerGroup(withNamespace(this.producerGroup)); this.setProducerGroup(withNamespace(this.producerGroup));
@ -19,7 +21,7 @@ org.apache.rocketmq.client.producer.DefaultMQProducer#start
第一步、检查 producerGroup 第一步、检查 producerGroup
``` ```java
private void checkConfig() throws MQClientException { private void checkConfig() throws MQClientException {
Validators.checkGroup(this.defaultMQProducer.getProducerGroup()); Validators.checkGroup(this.defaultMQProducer.getProducerGroup());
@ -35,7 +37,7 @@ private void checkConfig() throws MQClientException {
第二步、设置 instanceName 第二步、设置 instanceName
``` ```java
public void changeInstanceNameToPID() { public void changeInstanceNameToPID() {
if (this.instanceName.equals("DEFAULT")) { if (this.instanceName.equals("DEFAULT")) {
this.instanceName = UtilAll.getPid() + "#" + System.nanoTime(); this.instanceName = UtilAll.getPid() + "#" + System.nanoTime();
@ -45,7 +47,7 @@ public void changeInstanceNameToPID() {
第三步、创建 mqClientInstance,它是与 nameserver 和 broker 通信的中介 第三步、创建 mqClientInstance,它是与 nameserver 和 broker 通信的中介
```jsx ```java
public MQClientInstance getOrCreateMQClientInstance(final ClientConfig clientConfig, RPCHook rpcHook) { public MQClientInstance getOrCreateMQClientInstance(final ClientConfig clientConfig, RPCHook rpcHook) {
String clientId = clientConfig.buildMQClientId(); String clientId = clientConfig.buildMQClientId();
MQClientInstance instance = this.factoryTable.get(clientId); MQClientInstance instance = this.factoryTable.get(clientId);
@ -68,7 +70,7 @@ public MQClientInstance getOrCreateMQClientInstance(final ClientConfig clientCon
第四步、将生产者加入 mqClientInstance 管理 第四步、将生产者加入 mqClientInstance 管理
``` ```java
public synchronized boolean registerProducer(final String group, final DefaultMQProducerImpl producer) { public synchronized boolean registerProducer(final String group, final DefaultMQProducerImpl producer) {
if (null == group || null == producer) { if (null == group || null == producer) {
return false; return false;
@ -86,9 +88,9 @@ public synchronized boolean registerProducer(final String group, final DefaultMQ
第五步、启动 MQClientInstance有一些关于消费者的任务 会在消费者启动流程中讲解) 第五步、启动 MQClientInstance有一些关于消费者的任务 会在消费者启动流程中讲解)
1> 启动netty客户端 创建与nameserver、broker通信的channel 1. 启动 netty 客户端 ,创建与 nameserver、broker 通信的 channel
``` ```java
public void start() { public void start() {
this.defaultEventExecutorGroup = new DefaultEventExecutorGroup( this.defaultEventExecutorGroup = new DefaultEventExecutorGroup(
nettyClientConfig.getClientWorkerThreads(), nettyClientConfig.getClientWorkerThreads(),
@ -158,11 +160,11 @@ public void start() {
} }
``` ```
2> 启动一些周期性的任务: 2. 启动一些周期性的任务:
更新 nameserver 地址的任务: 更新 nameserver 地址的任务:
``` ```java
if (null == this.clientConfig.getNamesrvAddr()) { if (null == this.clientConfig.getNamesrvAddr()) {
this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() { this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
@ -180,7 +182,7 @@ if (null == this.clientConfig.getNamesrvAddr()) {
更新 topic 路由信息的任务: 更新 topic 路由信息的任务:
``` ```java
this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() { this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
@Override @Override
@ -196,7 +198,7 @@ this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
更新 broker 的任务: 更新 broker 的任务:
``` ```java
this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() { this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
@Override @Override
@ -215,7 +217,7 @@ this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
`this.pullMessageService.start();` `this.pullMessageService.start();`
``` ```java
public void run() { public void run() {
log.info(this.getServiceName() + " service started"); log.info(this.getServiceName() + " service started");

Loading…
Cancel
Save