4、RocketMQ中Producer的启动（四）

青旅半醒 2024-02-05 12:12 13阅读 0赞

在NameServer与Broker启动之后，我们就可以来创建Producer进行生产消息，客户端常用的生产类是DefaultMQProducer，我们启动消费者其实就是调用该类的start方法。  
![在这里插入图片描述][d715b6137c904854bba472d62906b5ab.png]  
初始化逻辑

通过构建一起DefaultMQProducer类来实现初始化，查看源码我们可以得到DefaultMQProducer类的构造方法存在很多类型，但是最终会去执行最后一个构造方法。

public DefaultMQProducer(final String namespace, final String producerGroup, RPCHook rpcHook) {
        
        // 命名空间
        this.namespace = namespace;
        // 生产者组
        this.producerGroup = producerGroup;
        // 创建DefaultMQProducerImpl实例，负责发送消息
        defaultMQProducerImpl = new DefaultMQProducerImpl(this, rpcHook);
    }

该构造函数主要就是用来指定命名空间，生产者组以及用来处理消息的发送的DefaultMQProducerImpl类，该类里面包含消息处理的所有方法。

public DefaultMQProducerImpl(final DefaultMQProducer defaultMQProducer, RPCHook rpcHook) {
        
        this.defaultMQProducer = defaultMQProducer;
        this.rpcHook = rpcHook;
      // 异步发送消息的线程池队列
        this.asyncSenderThreadPoolQueue = new LinkedBlockingQueue<Runnable>(50000);
        // 初始化线程
        this.defaultAsyncSenderExecutor = new ThreadPoolExecutor(
            Runtime.getRuntime().availableProcessors(),
            Runtime.getRuntime().availableProcessors(),
            1000 * 60,
            TimeUnit.MILLISECONDS,
            this.asyncSenderThreadPoolQueue,
            new ThreadFactory() {
        
                private AtomicInteger threadIndex = new AtomicInteger(0);
    
                @Override
                public Thread newThread(Runnable r) {
        
                    return new Thread(r, "AsyncSenderExecutor_" + this.threadIndex.incrementAndGet());
                }
            });
    }

主要就是初始化一个异步发送消息的线程池。

启动逻辑

生产者的启动逻辑调用start方法，其实就是去调用DefaultMQProducerImpl的start方法。

public void start(final boolean startFactory) throws MQClientException {
        
        // 根据serviceState的状态来判断是否重复启动
        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();
                }
                // 客户端核心管理组件的初始化mQClientFactory
                this.mQClientFactory = MQClientManager.getInstance().getOrCreateMQClientInstance(this.defaultMQProducer, rpcHook);
                // 往核心组件中注册一个Producer，往hashMap插入数据
                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);
                }
                // 注册topicPublishInfoTable
                this.topicPublishInfoTable.put(this.defaultMQProducer.getCreateTopicKey(), new TopicPublishInfo());
    
                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;
        }
    
        this.mQClientFactory.sendHeartbeatToAllBrokerWithLock();
    
        RequestFutureHolder.getInstance().startScheduledTask(this);
    }

首先使用checkConfig去检查配置信息，比如生产者组等是否符合规范，并通过getOrCreateMQClientInstance方法根据clientId获取mQClientFactory的实例。

将生产者注册到MQClientInstance实例的ProducerTable当中，同时注册注册topicPublishInfoTable对象。

启动MQClientInstance实例，调用其start方法用于初始化netty服务，定时任务，拉取消息服务等。

最后主动调用sendHeartbeatToAllBrokerWithLock方法发送心跳信息给所有Broker并且移除超时的request请求，执行异常回调。

MQClientInstance的启动

MQClientInstance类封装了RockerMQ底层通讯处理的API，Producer与Consumer类都会使用到这个类，它是Producer，Consumer与NameServer，Broker打交道的网络通道，这个类可以理解成一个工厂，是对消费者与生产者以及控制台三者的一个合集，内部封装了netty客户端，消息的生成，消费和负载均衡的实现类等。

MQClientInstance类的初始化

public MQClientInstance(ClientConfig clientConfig, int instanceIndex, String clientId, RPCHook rpcHook) {
        
        this.clientConfig = clientConfig;
        this.instanceIndex = instanceIndex;
        // 创建netty客户端配置类
        this.nettyClientConfig = new NettyClientConfig();
        this.nettyClientConfig.setClientCallbackExecutorThreads(clientConfig.getClientCallbackExecutorThreads());
        this.nettyClientConfig.setUseTLS(clientConfig.isUseTLS());
        // 客户端请求处理器
        this.clientRemotingProcessor = new ClientRemotingProcessor(this);
        // 创建客户端远程通信API实现类的实例
        this.mQClientAPIImpl = new MQClientAPIImpl(this.nettyClientConfig, this.clientRemotingProcessor, rpcHook, clientConfig);
    
        // 更新nameServer的地址
        if (this.clientConfig.getNamesrvAddr() != null) {
        
            this.mQClientAPIImpl.updateNameServerAddressList(this.clientConfig.getNamesrvAddr());
            log.info("user specified name server address: {}", this.clientConfig.getNamesrvAddr());
        }
      // 客户端ID
        this.clientId = clientId;
      // mq的admin控制台操作实现
        this.mQAdminImpl = new MQAdminImpl(this);
      // push模式下拉取消息服务
        this.pullMessageService = new PullMessageService(this);
      // 负载均衡服务器
        this.rebalanceService = new RebalanceService(this);
    
        this.defaultMQProducer = new DefaultMQProducer(MixAll.CLIENT_INNER_PRODUCER_GROUP);
        this.defaultMQProducer.resetClientConfig(clientConfig);
      // 消费者统计管理器
        this.consumerStatsManager = new ConsumerStatsManager(this.scheduledExecutorService);
    
        log.info("Created a new client Instance, InstanceIndex:{}, ClientID:{}, ClientConfig:{}, ClientVersion:{}, SerializerType:{}",
            this.instanceIndex,
            this.clientId,
            this.clientConfig,
            MQVersion.getVersionDesc(MQVersion.CURRENT_VERSION), RemotingCommand.getSerializeTypeConfigInThisServer());
    }

初始化MQClientInstance类的时候会初始化netty客户端以及各种服务实例等。

MQClientInstance类的启动  
在调用MQClientInstance的start方法时，会启动很多相关的类比如初始化netty客户端等，接下来我们就一起来分析。

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()) {
        
                        // 获取nameServer的地址
                        this.mQClientAPIImpl.fetchNameServerAddr();
                    }
                    // Start request-response channel+
                    // 启动mQClientAPIImpl,建立客户端与服务端的channel
                    this.mQClientAPIImpl.start();
                    // Start various schedule tasks
                    // 启动定时任务
                    this.startScheduledTask();
                    // Start pull service
                    // 启动拉消息服务
                    // 发送消息
                    this.pullMessageService.start();
                    // Start rebalance service
                    // 负载均衡 rebalanceService为线程类，start方法执行run方法
                    // 负载均衡服务，消费者与生产者建立关系
                    this.rebalanceService.start();
                    // Start push service
                    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;
            }
        }
    }

首先会去启动一个netty的客户端，在前面分析的NettyRemotingServer类里面有过类似的启动逻辑，这里就不再重复，感兴趣的客户去看一下。

调用this.startScheduledTask()启动很多的定时任务，获取nameServer地址的，更新topic路由的，清除取消Broker以及发送心跳信息到所有Broker，持久化消费者进度，广播消息持久化到本地，集群消息推送到Broker端，尝试调整消费线程池的线程数量(目前还未实现该方法)。

注意：该类为生产者和消费者公用的类，所以作用于每一个Producer与Consumer。

private void startScheduledTask() {
        
        if (null == this.clientConfig.getNamesrvAddr()) {
        
            // 定时获取nameServer的地址
            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);
        }
        // 更新topic路由
        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.getPollNameServerInterval(), TimeUnit.MILLISECONDS);
        // 清理无效的Broker以及发送心跳信息给所有的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);
        // 持久化消费者偏移量，即消费进度 广播消息持久化到本地，集群消息推送到Broker端
        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);
        // 尝试调整push模式的消费线程池的线程数量（没有实现逻辑）
        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);
    }

更新topic路由信息

每隔30s从nameServer内部拉取并更新topic路由消息。

public void updateTopicRouteInfoFromNameServer() {
        
        // 使用set集合就是为了有效的去重
        Set<String> topicList = new HashSet<String>();
    
        // 从ConsumerTable中获取
        {
        
            Iterator<Entry<String, MQConsumerInner>> it = this.consumerTable.entrySet().iterator();
            while (it.hasNext()) {
        
                Entry<String, MQConsumerInner> entry = it.next();
                MQConsumerInner impl = entry.getValue();
                if (impl != null) {
        
                    Set<SubscriptionData> subList = impl.subscriptions();
                    if (subList != null) {
        
                        for (SubscriptionData subData : subList) {
        
                            topicList.add(subData.getTopic());
                        }
                    }
                }
            }
        }
    
        // 从producerTable中获取
        {
        
            Iterator<Entry<String, MQProducerInner>> it = this.producerTable.entrySet().iterator();
            while (it.hasNext()) {
        
                Entry<String, MQProducerInner> entry = it.next();
                MQProducerInner impl = entry.getValue();
                if (impl != null) {
        
                    Set<String> lst = impl.getPublishTopicList();
                    topicList.addAll(lst);
                }
            }
        }
      // 比较更新本地的路由信息
        for (String topic : topicList) {
        
         /**
         * 从nameSerer拉取到topic路由信息之后，调用topicRouteDataIsChange方法与本地
         * 的旧topic路由信息比较看是否更改，比较的数据包括：topic的队列信息queueDatas，
         * topic的broker信息brokerDatas，顺序topic配置orderTopicConf，
         * 消费过滤信息filterServerTable。
         * 当判断需要更新的时候，会更新本地的topic缓存，包括：
         *      1. 更新brokerName到brokerAddr的地址的映射关系，即brokerAddrTable；
         *      2. 更新生产者producerTable集合，更新MQProducerInner的topicPublishInfoTable属性；
         *      3. 更新消费者的consumerTable集合，更新MQConsumerInner的rebalanceImpl.topicSubscribeInfoTable属性；
         *      4. 更新topicRouteTable集合，更新本地topic路由信息。
       */
            this.updateTopicRouteInfoFromNameServer(topic);
        }
    }

主要逻辑就是从consumerTable以及producerTable中获取配置的所有topic集合，包括consumer订阅的topic集合以及Producer中topicPublishTable集合中的数据。

从nameSerer拉取到topic路由信息之后，调用topicRouteDataIsChange方法与本地的旧topic路由信息比较看是否更改，比较的数据包括：topic的队列信息queueDatas，topic的broker信息brokerDatas，顺序topic配置orderTopicConf，消费过滤信息filterServerTable。

清理无效的Broker

调用cleanOfflineBroker()方法区清除下线的Broker。

private void cleanOfflineBroker() {
        
        try {
        
            // 加锁
            if (this.lockNamesrv.tryLock(LOCK_TIMEOUT_MILLIS, TimeUnit.MILLISECONDS))
                try {
        
                    ConcurrentHashMap<String, HashMap<Long, String>> updatedTable = new ConcurrentHashMap<String, HashMap<Long, String>>(this.brokerAddrTable.size(), 1);
    
                    Iterator<Entry<String, HashMap<Long, String>>> itBrokerTable = this.brokerAddrTable.entrySet().iterator();
                    // 遍历brokerAddrTable
                    while (itBrokerTable.hasNext()) {
        
                        Entry<String, HashMap<Long, String>> entry = itBrokerTable.next();
                        String brokerName = entry.getKey();
                        HashMap<Long, String> oneTable = entry.getValue();
    
                        HashMap<Long, String> cloneAddrTable = new HashMap<Long, String>(oneTable.size(), 1);
                        cloneAddrTable.putAll(oneTable);
    
                        Iterator<Entry<Long, String>> it = cloneAddrTable.entrySet().iterator();
                        while (it.hasNext()) {
        
                            Entry<Long, String> ee = it.next();
                            String addr = ee.getValue();
                             // 判断broker地址是否存在于topicRouteTable的任意一个topic的路由信息中，
                            // 如果不存在则直接移除该broker地址
                            if (!this.isBrokerAddrExistInTopicRouteTable(addr)) {
        
                                it.remove();
                                log.info("the broker addr[{} {}] is offline, remove it", brokerName, addr);
                            }
                        }
                      // 集合为空就移除
                        if (cloneAddrTable.isEmpty()) {
        
                            itBrokerTable.remove();
                            log.info("the broker[{}] name's host is offline, remove it", brokerName);
                        } else {
        
                            // 否则更新剩下的
                            updatedTable.put(brokerName, cloneAddrTable);
                        }
                    }
                  // 不为空直接更新
                    if (!updatedTable.isEmpty()) {
        
                        this.brokerAddrTable.putAll(updatedTable);
                    }
                } finally {
        
                    this.lockNamesrv.unlock();
                }
        } catch (InterruptedException e) {
        
            log.warn("cleanOfflineBroker Exception", e);
        }
    }

该方法间隔30s就会被执行，遍历并更新brokerAddrTable集合，其主要的步骤就是获取每一个人对象地址，首先去本地topicRouteTable是否存在，存在即保留不存在就表名Broker以下线需要被清除，如果brokerAddrTable中的value集合也是空则会直接删除键值对。

发送心跳信息给Broker

该方法每隔30s向所有的Broker发送心跳包的定时任务方法，客户的consumer和producer都是通过该定时任务发送心跳数据包的。在其他地方也会主动调用一次该方法，例如DefaultMQProducerImpl、DefaultMQPushConsumerImpl等类的start方法的结尾都会主动调用一次该方法。

public void sendHeartbeatToAllBrokerWithLock() {
        
         // 加锁
        if (this.lockHeartbeat.tryLock()) {
        
            try {
        
                // 发送心跳包给所有broker
                this.sendHeartbeatToAllBroker();
                // 上传过滤类到Broker对应的所有Filtersrv，push模式消费使用
                this.uploadFilterClassSource();
            } catch (final Exception e) {
        
                log.error("sendHeartbeatToAllBroker exception", e);
            } finally {
        
                this.lockHeartbeat.unlock();
            }
        } else {
        
            log.warn("lock heartBeat, but failed. [{}]", this.clientId);
        }
    }

首先会通过prepareHeartbeatData方法准备心跳包的数据进行发送，然后会遍历brokerAddrTable的broker地址进行循环发送心跳包。  
注意：如果单单启动了生产者，这时候我们要去判断是否为Master节点，由于生产者只能与Master发送消息数据，而当我们启动消费者此时我们就需要向所有的broker发送心跳包。

持久化消费偏移量

@Override
    public void persistConsumerOffset() {
        
        try {
        
            this.makeSureStateOK();
            Set<MessageQueue> mqs = new HashSet<MessageQueue>();
            Set<MessageQueue> allocateMq = this.rebalanceImpl.getProcessQueueTable().keySet();
            mqs.addAll(allocateMq);
          // 持久化的方法
            this.offsetStore.persistAll(mqs);
        } catch (Exception e) {
        
            log.error("group: " + this.defaultMQPushConsumer.getConsumerGroup() + " persistConsumerOffset exception", e);
        }
    }

调用persistAllConsumerOffset()进行持久化，而持久化的功能时消费端所有的。通过调用DefaultMQPushConsumer的persistAll()方法，如果你是广播就会执行LocalFileOffsetStore的对应方法，如果你是集群就会执行RemoteBrokerOffsetStore的对应方法。

发送消息的逻辑在Producer的消息发送逻辑里面，下一章分析。

负载均衡也是相对于消费端，消费端在处理消息的时候使用的消费模式，后期去消费端里面详解。

@Override
    public void run() {
        
        log.info(this.getServiceName() + " service started");
    
        while (!this.isStopped()) {
        
            this.waitForRunning(waitInterval);
            // 执行该方法进行负载均衡
            this.mqClientFactory.doRebalance();
        }
    
        log.info(this.getServiceName() + " service end");
    }

[d715b6137c904854bba472d62906b5ab.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/02/05/4ca582f950ab4ce293deaedcfc50dea8.png