RocketMQ探索-Producer的start

系统管理员 2022-01-14 03:11 266阅读 0赞

[2019独角兽企业重金招聘Python工程师标准>>> ][2019_Python_] ![hot3.png][]

在RocketMQ中，使用Producer相关类来生产消息，第一次使用的时，会调用producer.start()方法来进行初始化。这里我们来探索一下Producer的start做了些什么。

时序图如下：

![165756\_9cl1\_3134950.png][165756_9cl1_3134950.png]

producer.start()其实是调用DefaultMQProducerImpl.start()，重点看看里面2句代码：

boolean registerOK = mQClientFactory.registerProducer(this.defaultMQProducer.getProducerGroup(), this);

说明：把DefaultMQProducer对象添加到MQClientInstance的producerTable属性中:

public boolean registerProducer(final String group, final DefaultMQProducerImpl producer) {
        if (null == group || null == producer) {
            return false;
        }
    
        MQProducerInner prev = this.producerTable.putIfAbsent(group, producer);
        if (prev != null) {
            log.warn("the producer group[{}] exist already.", group);
            return false;
        }
    
        return true;
    }

producerTable对象里面存储producerGroupName和DefaultMQProducer的映射。key-value:<producerGroupName,DefaultMQProducer>

if (startFactory) {
        mQClientFactory.start();
    }

mQClientFactory.start()调用MQClientInstance（负责启动通信服务和定时任务）的start：

public void start() throws MQClientException {
    
        synchronized (this) {
            switch (this.serviceState) {
                case CREATE_JUST:
                    this.serviceState = ServiceState.START_FAILED;
                    // If not specified,looking address from name server
                    if (null == this.clientConfig.getNamesrvAddr()) {
                        this.clientConfig.setNamesrvAddr(this.mQClientAPIImpl.fetchNameServerAddr());
                    }
                    // Start request-response channel 启动MQClientAPIImpl
                    this.mQClientAPIImpl.start();
                    // Start various schedule tasks
                    this.startScheduledTask();
                    // Start pull service
                    this.pullMessageService.start();
                    // Start rebalance service
                    this.rebalanceService.start();
                    // Start push service 用于将消费失败的消息发回broker
                    this.defaultMQProducer.getDefaultMQProducerImpl().start(false);
                    log.info("the client factory [{}] start OK", this.clientId);
                    this.serviceState = ServiceState.RUNNING;
                    break;
                case RUNNING:
                    break;
                case SHUTDOWN_ALREADY:
                    break;
                case START_FAILED:
                    throw new MQClientException("The Factory object[" + this.getClientId() + "] has been created before, and failed.", null);
                default:
                    break;
            }
        }
    }

在Producer启动的时候，serviceState值为CREATE\_JUST。因为在以上的方法中就是调用各种的start方法：

*      this.mQClientAPIImpl.start();

说明：调用：MQClientAPIImpl.start()--->NettyRemotingClient.start()：MQClientAPIImpl(客户端与远程交互的封装,其内部使用了RemotingClient来实现与远程的交互),NettyRemotingClient.start()的方法里面启动了netty的通信客户端（这里就不对netty做介绍了，因为这个是个大工程）。

*      this.startScheduledTask();

说明：启动各种的定时任务：

*      /**
         * 启动各种定时任务
         */
        private void startScheduledTask() {
            //每两分钟执行一次寻址服务(NameServer地址)
            if (null == this.clientConfig.getNamesrvAddr()) {
                this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
        
                    @Override
                    public void run() {
                        try {
                            MQClientInstance.this.mQClientAPIImpl.fetchNameServerAddr();
                        } catch (Exception e) {
                            log.error("ScheduledTask fetchNameServerAddr exception", e);
                        }
                    }
                }, 1000 * 10, 1000 * 60 * 2, TimeUnit.MILLISECONDS);
            }
        
            //每30秒更新一次所有的topic的路由信息(topicRouteTable)
            this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
        
                @Override
                public void run() {
                    try {
                        MQClientInstance.this.updateTopicRouteInfoFromNameServer();
                    } catch (Exception e) {
                        log.error("ScheduledTask updateTopicRouteInfoFromNameServer exception", e);
                    }
                }
            }, 10, this.clientConfig.getPollNameServerInteval(), TimeUnit.MILLISECONDS);
        
            //每30秒移除离线的broker
            //每30秒发送一次心跳给所有的master broker
            this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
        
                @Override
                public void run() {
                    try {
                        MQClientInstance.this.cleanOfflineBroker();
                        MQClientInstance.this.sendHeartbeatToAllBrokerWithLock();
                    } catch (Exception e) {
                        log.error("ScheduledTask sendHeartbeatToAllBroker exception", e);
                    }
                }
            }, 1000, this.clientConfig.getHeartbeatBrokerInterval(), TimeUnit.MILLISECONDS);
        
            //更新offset每5秒提交一次消费的offset，broker端为ConsumerOffsetManager负责记录，此offset是逻辑偏移量，比如说，consumerA@consumerAGroup 在broker_a的queue 0的消费队列共有10000条消息，目前消费到888，那么offset就是888.
            //因为producer和consumer内部都持有MQClientInstance实例，故MQClientInstance既有生产者逻辑，又有消费者逻辑。
            this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
        
                @Override
                public void run() {
                    try {
                        MQClientInstance.this.persistAllConsumerOffset();
                    } catch (Exception e) {
                        log.error("ScheduledTask persistAllConsumerOffset exception", e);
                    }
                }
            }, 1000 * 10, this.clientConfig.getPersistConsumerOffsetInterval(), TimeUnit.MILLISECONDS);
        
            //每1分钟调整一次线程池，这也是针对消费者来说的，具体为如果消息堆积超过10W条，则调大线程池，最多64个线程；如果消息堆积少于8W条，则调小线程池，最少20的线程。
            this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
        
                @Override
                public void run() {
                    try {
                        MQClientInstance.this.adjustThreadPool();
                    } catch (Exception e) {
                        log.error("ScheduledTask adjustThreadPool exception", e);
                    }
                }
            }, 1, 1, TimeUnit.MINUTES);
        }

查看方法上面的注释。

*      this.pullMessageService.start();

说明：consumer的拉取消息线程实现方式：PullMessageService继承ServiceThread（对拉取消息请求进行了封装，使其队列化），start拉取消息线程启动，在run方法里面实现了：不断的从pullRequestQueue中取出请求，并调用消息拉取。

*      this.rebalanceService.start();

说明：再平衡线程启动

*      this.defaultMQProducer.getDefaultMQProducerImpl().start(false);

说明：暂时没有搞明白

总结：在Producer启动的时候，做了几件事情：

1.  把自己保持到producerTable属性中<groupName,Producer>
2.  启动各种定时任务（对MQClientInstance说明一下，因为生产和消费者都会持有MQClientInstance所以在启动任务的时候会启动生产和消费相关的任务线程）

启动执行寻址服务的任务（NameServer地址）。

启动更新所有的topic的路由信息(topicRouteTable)的任务。

启动移除离线的broker和发送心跳给所有的master broker的任务。

启动提交消费的offset（逻辑偏移量）到broker（broker端为ConsumerOffsetManager负责记录） 的任务。

启动调整线程池的任务（针对消费）。

转载于:https://my.oschina.net/u/3134950/blog/1456821

[2019_Python_]: https://my.oschina.net/u/2663968/blog/3061697
[hot3.png]: /images/20220113/fdf1e41e5c57488aa5231f54238e4c1b.png
[165756_9cl1_3134950.png]: https://static.oschina.net/uploads/space/2017/0717/165756_9cl1_3134950.png