快捷導(dǎo)航

RocketMQ?producer發(fā)送者淺析

更新時(shí)間：2023年04月24日 11:33:09 作者：Acqierement

RocketMQ生產(chǎn)者是一種高性能、可靠的消息發(fā)送者，能夠?qū)⑾⒖焖?、可靠地發(fā)送到RocketMQ消息隊(duì)列中。它具有多種消息發(fā)送模式和消息發(fā)送方式，可以根據(jù)不同的業(yè)務(wù)需求進(jìn)行靈活配置

發(fā)送者其實(shí)比較簡(jiǎn)單，需要做的就是首先確定往哪里發(fā)送，其次怎么讓消息發(fā)送順暢。我們就看一下具體的代碼吧。

首先調(diào)用start方法。完成各個(gè)類的初始化，啟動(dòng)多個(gè)定時(shí)任務(wù)，其中一個(gè)定時(shí)任務(wù)是updateTopicRouteInfoFromNameServer，這個(gè)方法里面和nameService建立長(zhǎng)連接，同時(shí)維護(hù)了topicRouteTable和brokerAddrTable等緩存。topicRouteTable里面維護(hù)了這個(gè)topic包括有哪些queue和broker。這樣producer才可以知道要發(fā)往哪里。

啟動(dòng)的流程主要在這個(gè)方法中：

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.mQClientAPIImpl.fetchNameServerAddr();
                }
                // Start request-response channel
                this.mQClientAPIImpl.start();
                // Start various schedule tasks
                this.startScheduledTask();
                // Start pull service
                this.pullMessageService.start();
                // Start rebalance service
                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;
        }
    }
}

其中啟動(dòng)了一系列定時(shí)任務(wù)，包括org.apache.rocketmq.client.impl.factory.MQClientInstance#updateTopicRouteInfoFromNameServer這個(gè)方法

    public boolean updateTopicRouteInfoFromNameServer(final String topic, boolean isDefault,
        DefaultMQProducer defaultMQProducer) {
        try {
            if (this.lockNamesrv.tryLock(LOCK_TIMEOUT_MILLIS, TimeUnit.MILLISECONDS)) {
                try {
                    TopicRouteData topicRouteData;
                    if (isDefault && defaultMQProducer != null) {
                        // 從nameServer獲取topciRouteData
                        topicRouteData = this.mQClientAPIImpl.getDefaultTopicRouteInfoFromNameServer(defaultMQProducer.getCreateTopicKey(),
                            clientConfig.getMqClientApiTimeout());
                        if (topicRouteData != null) {
                            for (QueueData data : topicRouteData.getQueueDatas()) {
                                int queueNums = Math.min(defaultMQProducer.getDefaultTopicQueueNums(), data.getReadQueueNums());
                                data.setReadQueueNums(queueNums);
                                data.setWriteQueueNums(queueNums);
                            }
                        }
                    } else {
                        topicRouteData = this.mQClientAPIImpl.getTopicRouteInfoFromNameServer(topic, clientConfig.getMqClientApiTimeout());
                    }
                    if (topicRouteData != null) {
                        TopicRouteData old = this.topicRouteTable.get(topic);
                        boolean changed = topicRouteData.topicRouteDataChanged(old);
                        if (!changed) {
                            changed = this.isNeedUpdateTopicRouteInfo(topic);
                        } else {
                            log.info("the topic[{}] route info changed, old[{}] ,new[{}]", topic, old, topicRouteData);
                        }
                        if (changed) {
                            for (BrokerData bd : topicRouteData.getBrokerDatas()) {
                                this.brokerAddrTable.put(bd.getBrokerName(), bd.getBrokerAddrs());
                            }
                            // Update endpoint map
                            {
                                ConcurrentMap<MessageQueue, String> mqEndPoints = topicRouteData2EndpointsForStaticTopic(topic, topicRouteData);
                                if (!mqEndPoints.isEmpty()) {
                                    topicEndPointsTable.put(topic, mqEndPoints);
                                }
                            }
                            // Update Pub info
                            {
                                // 生成topicPublishInfo
                                TopicPublishInfo publishInfo = topicRouteData2TopicPublishInfo(topic, topicRouteData);
                                publishInfo.setHaveTopicRouterInfo(true);
                                for (Entry<String, MQProducerInner> entry : this.producerTable.entrySet()) {
                                    MQProducerInner impl = entry.getValue();
                                    if (impl != null) {
                                        // 更新 topicPublishInfo
                                        impl.updateTopicPublishInfo(topic, publishInfo);
                                    }
                                }
                            }
                            // Update sub info
                            if (!consumerTable.isEmpty()) {
                                Set<MessageQueue> subscribeInfo = topicRouteData2TopicSubscribeInfo(topic, topicRouteData);
                                for (Entry<String, MQConsumerInner> entry : this.consumerTable.entrySet()) {
                                    MQConsumerInner impl = entry.getValue();
                                    if (impl != null) {
                                        impl.updateTopicSubscribeInfo(topic, subscribeInfo);
                                    }
                                }
                            }
                            TopicRouteData cloneTopicRouteData = new TopicRouteData(topicRouteData);
                            log.info("topicRouteTable.put. Topic = {}, TopicRouteData[{}]", topic, cloneTopicRouteData);
                            this.topicRouteTable.put(topic, cloneTopicRouteData);
                            return true;
                        }
                    } else {
                        log.warn("updateTopicRouteInfoFromNameServer, getTopicRouteInfoFromNameServer return null, Topic: {}. [{}]", topic, this.clientId);
                    }
                } catch (MQClientException e) {
                    if (!topic.startsWith(MixAll.RETRY_GROUP_TOPIC_PREFIX) && !topic.equals(TopicValidator.AUTO_CREATE_TOPIC_KEY_TOPIC)) {
                        log.warn("updateTopicRouteInfoFromNameServer Exception", e);
                    }
                } catch (RemotingException e) {
                    log.error("updateTopicRouteInfoFromNameServer Exception", e);
                    throw new IllegalStateException(e);
                } finally {
                    this.lockNamesrv.unlock();
                }
            } else {
                log.warn("updateTopicRouteInfoFromNameServer tryLock timeout {}ms. [{}]", LOCK_TIMEOUT_MILLIS, this.clientId);
            }
        } catch (InterruptedException e) {
            log.warn("updateTopicRouteInfoFromNameServer Exception", e);
        }
        return false;
    }

通過方法名也知道是從nameServer獲取這個(gè)topic相關(guān)的broke數(shù)據(jù)，拿到TopicRouteData數(shù)據(jù)。先更新brokerAddrTable，存儲(chǔ)borker具體的地址。然后在org.apache.rocketmq.client.impl.factory.MQClientInstance#topicRouteData2TopicPublishInfo里面再進(jìn)一步生成TopicPublishInfo數(shù)據(jù)。TopicPublishInfo是對(duì)TopicRouteData的一個(gè)封裝，除了TopicRouteData，還有messageQueue數(shù)據(jù)，messageQueue是Queue和Borker的交集，會(huì)根據(jù)配置的queue數(shù)量，生成具體的messageQueue，queueId就是0，1，2，3，4他們自己的順序。

所以有了TopicPublishInfo數(shù)據(jù)，就知道往哪里發(fā)了。

發(fā)送消息的過程。

先找到TopicPublishInfo。TopicPublishInfo里面有一個(gè)MessageQueue的list。
從MessageQueueList里面拿到一個(gè)messageQueue。如果沒有開啟sendLatencyFaultEnable，默認(rèn)就是采用輪詢方法。具體的輪詢方式就是，TopicPublishInfo里面維護(hù)了一個(gè)序號(hào)index，每次index自增1，然后通過index去MessageQueueList里面拿一個(gè)。
拿到了MessageQueue之后，里面有broker的name，根據(jù)name去找broker的ip地址，發(fā)送數(shù)據(jù)。這個(gè)ip地址就是前面提到的brokerAddrTable變量，在updateTopicRouteInfoFromNameServer方法里面維護(hù)的。

到此這篇關(guān)于RocketMQ producer發(fā)送者淺析的文章就介紹到這了,更多相關(guān)RocketMQ producer內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: