RocketMQ的前提回顧

RocketMQ是一款分布式、隊(duì)列模型的消息中間件，具有以下特點(diǎn)：

1. 能夠保證嚴(yán)格的消息順序
2. 提供豐富的消息拉取模式
3. 高效的訂閱者水平擴(kuò)展能力
4. 實(shí)時(shí)的消息訂閱機(jī)制
5. 億級消息堆積能力

為什么使用RocketMQ

1. 強(qiáng)調(diào)集群無單點(diǎn)，可擴(kuò)展，任意一點(diǎn)高可用、水平可擴(kuò)展
2. 海量消息堆積能力，消息堆積后寫入低延遲
3. 支持上萬個(gè)隊(duì)列
4. 消息失敗重試機(jī)制
5. 消息可查詢
6. 開源社區(qū)活躍
7. 成熟度已經(jīng)經(jīng)過淘寶雙十一的考驗(yàn)

RocketMQ的發(fā)展變化

RocketMQ開源是使用文件作為持久化工具，阿里內(nèi)部未開源的性能會更高，使用oceanBase作為持久化工具。
在RocketMQ1.x和2.x使用zookeeper管理集群，3.x開始使用nameserver代替zk，更輕量級，此外RocketMQ的客戶端擁有兩種的操作方式：DefaultMQPushConsumer和DefaultMQPullConsumer。

DefaultMQPushConsumer的Maven配置

DefaultMQPushConsumer使用示例

1. CONSUME_FROM_LAST_OFFSET：第一次啟動(dòng)從隊(duì)列最后位置消費(fèi)，后續(xù)再啟動(dòng)接著上次消費(fèi)的進(jìn)度開始消費(fèi)
2. CONSUME_FROM_FIRST_OFFSET：第一次啟動(dòng)從隊(duì)列初始位置消費(fèi)，后續(xù)再啟動(dòng)接著上次消費(fèi)的進(jìn)度開始消費(fèi)
3. CONSUME_FROM_TIMESTAMP：第一次啟動(dòng)從指定時(shí)間點(diǎn)位置消費(fèi)，后續(xù)再啟動(dòng)接著上次消費(fèi)的進(jìn)度開始消費(fèi)

以上所說的第一次啟動(dòng)是指從來沒有消費(fèi)過的消費(fèi)者，如果該消費(fèi)者消費(fèi)過，那么會在broker端記錄該消費(fèi)者的消費(fèi)位置，如果該消費(fèi)者掛了再啟動(dòng)，那么自動(dòng)從上次消費(fèi)的進(jìn)度開始

public class MQPushConsumer {    public static void main(String[] args) throws MQClientException {        String groupName = "rocketMqGroup1";        // 用于把多個(gè)Consumer組織到一起，提高并發(fā)處理能力        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(groupName);        // 設(shè)置nameServer地址，多個(gè)以;分隔        consumer.setNamesrvAddr("name-serverl-ip:9876;name-server2-ip:9876"); consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);        consumer.setMessageModel(MessageModel.BROADCASTING);        // 訂閱topic，可以對指定消息進(jìn)行過濾，例如："TopicTest","tagl||tag2||tag3",*或null表示topic所有消息        consumer.subscribe("order-topic", "*");        consumer.registerMessageListener(new MessageListenerConcurrently() {            @Override            public ConsumeConcurrentlyStatus consumeMessage(List

• CLUSTERING：默認(rèn)模式，同一個(gè)ConsumerGroup(groupName相同)每個(gè)consumer只消費(fèi)所訂閱消息的一部分內(nèi)容，同一個(gè)ConsumerGroup里所有的Consumer消息加起來才是所
• 訂閱topic整體，從而達(dá)到負(fù)載均衡的目的
• BROADCASTING：同一個(gè)ConsumerGroup每個(gè)consumer都消費(fèi)到所訂閱topic所有消息，也就是一個(gè)消費(fèi)會被多次分發(fā)，被多個(gè)consumer消費(fèi)。

ConsumeConcurrentlyStatus.RECONSUME_LATER boker會根據(jù)設(shè)置的messageDelayLevel發(fā)起重試，默認(rèn)16次。

DefaultMQPushConsumerImpl中各個(gè)對象的主要功能如下：

RebalancePushImpl：主要負(fù)責(zé)決定，當(dāng)前的consumer應(yīng)該從哪些Queue中消費(fèi)消息；

• 1）PullAPIWrapper：長連接，負(fù)責(zé)從broker處拉取消息，然后利用ConsumeMessageService回調(diào)用戶的Listener執(zhí)行消息消費(fèi)邏輯；
• 2）ConsumeMessageService：實(shí)現(xiàn)所謂的"Push-被動(dòng)"消費(fèi)機(jī)制；從Broker拉取的消息后，封裝成ConsumeRequest提交給ConsumeMessageSerivce，此service負(fù)責(zé)回調(diào)用戶的Listener消費(fèi)消息；
• 3）OffsetStore：維護(hù)當(dāng)前consumer的消費(fèi)記錄（offset）；有兩種實(shí)現(xiàn)，Local和Rmote，Local存儲在本地磁盤上，適用于BROADCASTING廣播消費(fèi)模式；而Remote則將消費(fèi)進(jìn)度存儲在Broker上，適用于CLUSTERING集群消費(fèi)模式；
• 4）MQClientFactory：負(fù)責(zé)管理client（consumer、producer），并提供多中功能接口供各個(gè)Service（Rebalance、PullMessage等）調(diào)用；大部分邏輯均在這個(gè)類中完成；

consumer.registerMessageListener執(zhí)行過程：

/**     * Register a callback to execute on message arrival for concurrent consuming.     * @param messageListener message handling callback.     */    @Override    public void registerMessageListener(MessageListenerConcurrently messageListener) {        this.messageListener = messageListener; this.defaultMQPushConsumerImpl.registerMessageListener(messageListener);    }

通過源碼可以看出主要實(shí)現(xiàn)過程在DefaultMQPushConsumerImpl類中consumer.start后調(diào)用DefaultMQPushConsumerImpl的同步start方法

public synchronized void start() throws MQClientException {        switch (this.serviceState) {            case CREATE_JUST:                log.info("the consumer [{}] start beginning. messageModel={}, isUnitMode={}", this.defaultMQPushConsumer.getConsumerGroup(),                    this.defaultMQPushConsumer.getMessageModel(), this.defaultMQPushConsumer.isUnitMode());                this.serviceState = ServiceState.START_FAILED;                this.checkConfig();                this.copySubscription();                if (this.defaultMQPushConsumer.getMessageModel() == MessageModel.CLUSTERING) {                    this.defaultMQPushConsumer.changeInstanceNameToPID();                }                this.mQClientFactory = MQClientManager.getInstance().getAndCreateMQClientInstance(this.defaultMQPushConsumer, this.rpcHook);                this.rebalanceImpl.setConsumerGroup(this.defaultMQPushConsumer.getConsumerGroup());                this.rebalanceImpl.setMessageModel(this.defaultMQPushConsumer.getMessageModel());                this.rebalanceImpl.setAllocateMessageQueueStrategy(this.defaultMQPushConsumer.getAllocateMessageQueueStrategy());                this.rebalanceImpl.setmQClientFactory(this.mQClientFactory);                this.pullAPIWrapper = new PullAPIWrapper(                    mQClientFactory,                    this.defaultMQPushConsumer.getConsumerGroup(), isUnitMode());                this.pullAPIWrapper.registerFilterMessageHook(filterMessageHookList);                if (this.defaultMQPushConsumer.getOffsetStore() != null) {                    this.offsetStore = this.defaultMQPushConsumer.getOffsetStore();                } else {                    switch (this.defaultMQPushConsumer.getMessageModel()) {                        case BROADCASTING:                            this.offsetStore = new LocalFileOffsetStore(this.mQClientFactory, this.defaultMQPushConsumer.getConsumerGroup());                            break;                        case CLUSTERING:                            this.offsetStore = new RemoteBrokerOffsetStore(this.mQClientFactory, this.defaultMQPushConsumer.getConsumerGroup());                            break;                        default:                            break;                    }                  this.defaultMQPushConsumer.setOffsetStore(this.offsetStore);                }                this.offsetStore.load();                if (this.getMessageListenerInner() instanceof MessageListenerOrderly) {                    this.consumeOrderly = true;                    this.consumeMessageService =                        new ConsumeMessageOrderlyService(this, (MessageListenerOrderly) this.getMessageListenerInner());                } else if (this.getMessageListenerInner() instanceof MessageListenerConcurrently) {                    this.consumeOrderly = false;                    this.consumeMessageService =                        new ConsumeMessageConcurrentlyService(this, (MessageListenerConcurrently) this.getMessageListenerInner());                }                this.consumeMessageService.start();                boolean registerOK = mQClientFactory.registerConsumer(this.defaultMQPushConsumer.getConsumerGroup(), this);                if (!registerOK) {                    this.serviceState = ServiceState.CREATE_JUST;                    this.consumeMessageService.shutdown();                    throw new MQClientException("The consumer group[" + this.defaultMQPushConsumer.getConsumerGroup()                        + "] has been created before, specify another name please." + FAQUrl.suggestTodo(FAQUrl.GROUP_NAME_DUPLICATE_URL),                        null);                }                mQClientFactory.start();                log.info("the consumer [{}] start OK.", this.defaultMQPushConsumer.getConsumerGroup());                this.serviceState = ServiceState.RUNNING;                break;            case RUNNING:            case START_FAILED:            case SHUTDOWN_ALREADY:                throw new MQClientException("The PushConsumer service state not OK, maybe started once, "                    + this.serviceState                    + FAQUrl.suggestTodo(FAQUrl.CLIENT_SERVICE_NOT_OK),                    null);            default:                break;        }        this.updateTopicSubscribeInfoWhenSubscriptionChanged();        this.mQClientFactory.checkClientInBroker();        this.mQClientFactory.sendHeartbeatToAllBrokerWithLock();        this.mQClientFactory.rebalanceImmediately();    }

通過mQClientFactory.start();發(fā)我們發(fā)現(xiàn)他調(diào)用

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 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;            }        }    }

在這個(gè)方法中有多個(gè)start，我們主要看pullMessageService.start();通過這里我們發(fā)現(xiàn)RocketMQ的Push模式底層其實(shí)也是通過pull實(shí)現(xiàn)的，下面我們來看下pullMessageService處理了哪些邏輯：

private void pullMessage(final PullRequest pullRequest) {        final MQConsumerInner consumer = this.mQClientFactory.selectConsumer(pullRequest.getConsumerGroup());        if (consumer != null) {            DefaultMQPushConsumerImpl impl = (DefaultMQPushConsumerImpl) consumer;            impl.pullMessage(pullRequest);        } else {            log.warn("No matched consumer for the PullRequest {}, drop it", pullRequest);        }    }

我們發(fā)現(xiàn)其實(shí)他還是通過DefaultMQPushConsumerImpl類的pullMessage方法來進(jìn)行消息的邏輯處理.

pullRequest拉取方式

PullRequest這里說明一下，上面我們已經(jīng)提了一下rocketmq的push模式其實(shí)是通過pull模式封裝實(shí)現(xiàn)的，pullrequest這里是通過長輪詢的方式達(dá)到push效果。

長輪詢方式既有pull的優(yōu)點(diǎn)又有push模式的實(shí)時(shí)性有點(diǎn)。

• push方式是server端接收到消息后，主動(dòng)把消息推送給client端，實(shí)時(shí)性高。弊端是server端工作量大，影響性能，其次是client端處理能力不同且client端的狀態(tài)不受server端的控制，如果client端不能及時(shí)處理消息容易導(dǎo)致消息堆積已經(jīng)影響正常業(yè)務(wù)等。

• pull方式是client循環(huán)從server端拉取消息，主動(dòng)權(quán)在client端，自己處理完一個(gè)消息再去拉取下一個(gè)，缺點(diǎn)是循環(huán)的時(shí)間不好設(shè)定，時(shí)間太短容易忙等，浪費(fèi)CPU資源，時(shí)間間隔太長client的處理能力會下降，有時(shí)候有些消息會處理不及時(shí)。

長輪詢的方式可以結(jié)合兩者優(yōu)點(diǎn)

1. 檢查PullRequest對象中的ProcessQueue對象的dropped是否為true（在RebalanceService線程中為topic下的MessageQueue創(chuàng)建拉取消息請求時(shí)要維護(hù)對應(yīng)的ProcessQueue對象，若Consumer不再訂閱該topic則會將該對象的dropped置為true）；若是則認(rèn)為該請求是已經(jīng)取消的，則直接跳出該方法；
2. 更新PullRequest對象中的ProcessQueue對象的時(shí)間戳（ProcessQueue.lastPullTimestamp）為當(dāng)前時(shí)間戳；
3. 檢查該Consumer是否運(yùn)行中，即DefaultMQPushConsumerImpl.serviceState是否為RUNNING;若不是運(yùn)行狀態(tài)或者是暫停狀態(tài)（DefaultMQPushConsumerImpl.pause=true），則調(diào)用PullMessageService.executePullRequestLater(PullRequest pullRequest, long timeDelay)方法延遲再拉取消息，其中timeDelay=3000；該方法的目的是在3秒之后再次將該P(yáng)ullRequest對象放入PullMessageService. pullRequestQueue隊(duì)列中；并跳出該方法；
4. 進(jìn)行流控。若ProcessQueue對象的msgCount大于了消費(fèi)端的流控閾值（DefaultMQPushConsumer.pullThresholdForQueue，默認(rèn)值為1000），則調(diào)用PullMessageService.executePullRequestLater方法，在50毫秒之后重新該P(yáng)ullRequest請求放入PullMessageService.pullRequestQueue隊(duì)列中；并跳出該方法；
5. 若不是順序消費(fèi)（即DefaultMQPushConsumerImpl.consumeOrderly等于false），則檢查ProcessQueue對象的msgTreeMap:TreeMap
6. 以PullRequest.messageQueue對象的topic值為參數(shù)從RebalanceImpl.subscriptionInner: ConcurrentHashMap, SubscriptionData>中獲取對應(yīng)的SubscriptionData對象，若該對象為null，考慮到并發(fā)的關(guān)系，調(diào)用executePullRequestLater方法，稍后重試；并跳出該方法；
7. 若消息模型為集群模式（RebalanceImpl.messageModel等于CLUSTERING），則以PullRequest對象的MessageQueue變量值、type =READ_FROM_MEMORY（從內(nèi)存中獲取消費(fèi)進(jìn)度offset值）為參數(shù)調(diào)用DefaultMQPushConsumerImpl. offsetStore對象（初始化為RemoteBrokerOffsetStore對象）的readOffset(MessageQueue mq, ReadOffsetType type)方法從本地內(nèi)存中獲取消費(fèi)進(jìn)度offset值。若該offset值大于0 則置臨時(shí)變量commitOffsetEnable等于true否則為false；該offset值作為pullKernelImpl方法中的commitOffset參數(shù)，在Broker端拉取消息之后根據(jù)commitOffsetEnable參數(shù)值決定是否用該offset更新消息進(jìn)度。該readOffset方法的邏輯是：以入?yún)essageQueue對象從RemoteBrokerOffsetStore.offsetTable:ConcurrentHashMap
8. 當(dāng)每次拉取消息之后需要更新訂閱關(guān)系（由DefaultMQPushConsumer. postSubscriptionWhenPull參數(shù)表示，默認(rèn)為false）并且以topic值參數(shù)從RebalanceImpl.subscriptionInner獲取的SubscriptionData對象的classFilterMode等于false（默認(rèn)為false），則將sysFlag標(biāo)記的第3個(gè)字節(jié)置為1，否則該字節(jié)置為0；
9. 該sysFlag標(biāo)記的第1個(gè)字節(jié)置為commitOffsetEnable的值；第2個(gè)字節(jié)（suspend標(biāo)記）置為1；第4個(gè)字節(jié)置為classFilterMode的值；
10. 初始化匿名內(nèi)部類PullCallback，實(shí)現(xiàn)了onSucess/onException方法； 該方法只有在異步請求的情況下才會回調(diào)；
11. 調(diào)用底層的拉取消息API接口：

PullAPIWrapper.pullKernelImpl

PullAPIWrapper.pullKernelImpl(MessageQueue mq, String subExpression, long subVersion,long offset, int maxNums, int sysFlag,long commitOffset,long brokerSuspendMaxTimeMillis, long timeoutMillis, CommunicationMode communicationMode, PullCallback pullCallback)方法進(jìn)行消息拉取操作。

將回調(diào)類PullCallback傳入該方法中，當(dāng)采用異步方式拉取消息時(shí)，在收到響應(yīng)之后會回調(diào)該回調(diào)類的方法。

public void pullMessage(final PullRequest pullRequest) {        final ProcessQueue processQueue = pullRequest.getProcessQueue();        if (processQueue.isDropped()) {            log.info("the pull request[{}] is dropped.", pullRequest.toString());            return;        }        pullRequest.getProcessQueue().setLastPullTimestamp(System.currentTimeMillis());        try {            this.makeSureStateOK();        } catch (MQClientException e) {            log.warn("pullMessage exception, consumer state not ok", e);            this.executePullRequestLater(pullRequest, PULL_TIME_DELAY_MILLS_WHEN_EXCEPTION);            return;        }        if (this.isPause()) {            log.warn("consumer was paused, execute pull request later. instanceName={}, group={}", this.defaultMQPushConsumer.getInstanceName(), this.defaultMQPushConsumer.getConsumerGroup());            this.executePullRequestLater(pullRequest, PULL_TIME_DELAY_MILLS_WHEN_SUSPEND);            return;        }        long cachedMessageCount = processQueue.getMsgCount().get();        long cachedMessageSizeInMiB = processQueue.getMsgSize().get() / (1024 * 1024);        if (cachedMessageCount > this.defaultMQPushConsumer.getPullThresholdForQueue()) {            this.executePullRequestLater(pullRequest, PULL_TIME_DELAY_MILLS_WHEN_FLOW_CONTROL);            if ((queueFlowControlTimes++ % 1000) == 0) {                log.warn(                    "the cached message count exceeds the threshold {}, so do flow control, minOffset={}, maxOffset={}, count={}, size={} MiB, pullRequest={}, flowControlTimes={}",                    this.defaultMQPushConsumer.getPullThresholdForQueue(), processQueue.getMsgTreeMap().firstKey(), processQueue.getMsgTreeMap().lastKey(), cachedMessageCount, cachedMessageSizeInMiB, pullRequest, queueFlowControlTimes);            }            return;        }        if (cachedMessageSizeInMiB > this.defaultMQPushConsumer.getPullThresholdSizeForQueue()) {            this.executePullRequestLater(pullRequest, PULL_TIME_DELAY_MILLS_WHEN_FLOW_CONTROL);            if ((queueFlowControlTimes++ % 1000) == 0) {                log.warn(                    "the cached message size exceeds the threshold {} MiB, so do flow control, minOffset={}, maxOffset={}, count={}, size={} MiB, pullRequest={}, flowControlTimes={}",                    this.defaultMQPushConsumer.getPullThresholdSizeForQueue(), processQueue.getMsgTreeMap().firstKey(), processQueue.getMsgTreeMap().lastKey(), cachedMessageCount, cachedMessageSizeInMiB, pullRequest, queueFlowControlTimes);            }            return;        }        if (!this.consumeOrderly) {            if (processQueue.getMaxSpan() > this.defaultMQPushConsumer.getConsumeConcurrentlyMaxSpan()) {                this.executePullRequestLater(pullRequest, PULL_TIME_DELAY_MILLS_WHEN_FLOW_CONTROL);                if ((queueMaxSpanFlowControlTimes++ % 1000) == 0) {                    log.warn(                        "the queues messages, span too long, so do flow control, minOffset={}, maxOffset={}, maxSpan={}, pullRequest={}, flowControlTimes={}",                        processQueue.getMsgTreeMap().firstKey(), processQueue.getMsgTreeMap().lastKey(), processQueue.getMaxSpan(),                        pullRequest, queueMaxSpanFlowControlTimes);                }                return;            }        } else {            if (processQueue.isLocked()) {                if (!pullRequest.isLockedFirst()) {                    final long offset = this.rebalanceImpl.computePullFromWhere(pullRequest.getMessageQueue());                    boolean brokerBusy = offset < pullRequest.getNextOffset();                    log.info("the first time to pull message, so fix offset from broker. pullRequest: {} NewOffset: {} brokerBusy: {}",                        pullRequest, offset, brokerBusy);                    if (brokerBusy) {                        log.info("[NOTIFYME]the first time to pull message, but pull request offset larger than broker consume offset. pullRequest: {} NewOffset: {}",                            pullRequest, offset);                    }                    pullRequest.setLockedFirst(true);                    pullRequest.setNextOffset(offset);                }            } else {                this.executePullRequestLater(pullRequest, PULL_TIME_DELAY_MILLS_WHEN_EXCEPTION);                log.info("pull message later because not locked in broker, {}", pullRequest);                return;            }        }        final SubscriptionData subscriptionData = this.rebalanceImpl.getSubscriptionInner().get(pullRequest.getMessageQueue().getTopic());        if (null == subscriptionData) {            this.executePullRequestLater(pullRequest, PULL_TIME_DELAY_MILLS_WHEN_EXCEPTION);            log.warn("find the consumers subscription failed, {}", pullRequest);            return;        }        final long beginTimestamp = System.currentTimeMillis();        PullCallback pullCallback = new PullCallback() {            @Override            public void onSuccess(PullResult pullResult) {                if (pullResult != null) {                    pullResult = DefaultMQPushConsumerImpl.this.pullAPIWrapper.processPullResult(pullRequest.getMessageQueue(), pullResult,                        subscriptionData);                    switch (pullResult.getPullStatus()) {                        case FOUND:                            long prevRequestOffset = pullRequest.getNextOffset();                            pullRequest.setNextOffset(pullResult.getNextBeginOffset());                            long pullRT = System.currentTimeMillis() - beginTimestamp;                            DefaultMQPushConsumerImpl.this.getConsumerStatsManager().incPullRT(pullRequest.getConsumerGroup(),                                pullRequest.getMessageQueue().getTopic(), pullRT);                            long firstMsgOffset = Long.MAX_VALUE;                            if (pullResult.getMsgFoundList() == null || pullResult.getMsgFoundList().isEmpty()) {                                DefaultMQPushConsumerImpl.this.executePullRequestImmediately(pullRequest);                            } else {                                firstMsgOffset = pullResult.getMsgFoundList().get(0).getQueueOffset();                                DefaultMQPushConsumerImpl.this.getConsumerStatsManager().incPullTPS(pullRequest.getConsumerGroup(),                                    pullRequest.getMessageQueue().getTopic(), pullResult.getMsgFoundList().size());                                boolean dispatchToConsume = processQueue.putMessage(pullResult.getMsgFoundList());                                DefaultMQPushConsumerImpl.this.consumeMessageService.submitConsumeRequest(                                    pullResult.getMsgFoundList(),                                    processQueue,                                    pullRequest.getMessageQueue(),                                    dispatchToConsume);                                if (DefaultMQPushConsumerImpl.this.defaultMQPushConsumer.getPullInterval() > 0) {                                    DefaultMQPushConsumerImpl.this.executePullRequestLater(pullRequest,                                        DefaultMQPushConsumerImpl.this.defaultMQPushConsumer.getPullInterval());                                } else {                                    DefaultMQPushConsumerImpl.this.executePullRequestImmediately(pullRequest);                                }                            }                            if (pullResult.getNextBeginOffset() < prevRequestOffset                                || firstMsgOffset < prevRequestOffset) {                                log.warn(                                    "[BUG] pull message result maybe data wrong, nextBeginOffset: {} firstMsgOffset: {} prevRequestOffset: {}",                                    pullResult.getNextBeginOffset(),                                    firstMsgOffset,                                    prevRequestOffset);                            }                            break;                        case NO_NEW_MSG:                            pullRequest.setNextOffset(pullResult.getNextBeginOffset());                            DefaultMQPushConsumerImpl.this.correctTagsOffset(pullRequest);                            DefaultMQPushConsumerImpl.this.executePullRequestImmediately(pullRequest);                            break;                        case NO_MATCHED_MSG:                            pullRequest.setNextOffset(pullResult.getNextBeginOffset());                            DefaultMQPushConsumerImpl.this.correctTagsOffset(pullRequest);                            DefaultMQPushConsumerImpl.this.executePullRequestImmediately(pullRequest);                            break;                        case OFFSET_ILLEGAL:                            log.warn("the pull request offset illegal, {} {}",                                pullRequest.toString(), pullResult.toString());                            pullRequest.setNextOffset(pullResult.getNextBeginOffset());                            pullRequest.getProcessQueue().setDropped(true);                            DefaultMQPushConsumerImpl.this.executeTaskLater(new Runnable() {                                @Override                                public void run() {                                    try {                                        DefaultMQPushConsumerImpl.this.offsetStore.updateOffset(pullRequest.getMessageQueue(),                                            pullRequest.getNextOffset(), false);                                        DefaultMQPushConsumerImpl.this.offsetStore.persist(pullRequest.getMessageQueue());                                        DefaultMQPushConsumerImpl.this.rebalanceImpl.removeProcessQueue(pullRequest.getMessageQueue());                                        log.warn("fix the pull request offset, {}", pullRequest);                                    } catch (Throwable e) {                                        log.error("executeTaskLater Exception", e);                                    }                                }                            }, 10000);                            break;                        default:                            break;                    }                }            }            @Override            public void onException(Throwable e) {                if (!pullRequest.getMessageQueue().getTopic().startsWith(MixAll.RETRY_GROUP_TOPIC_PREFIX)) {                    log.warn("execute the pull request exception", e);                }                DefaultMQPushConsumerImpl.this.executePullRequestLater(pullRequest, PULL_TIME_DELAY_MILLS_WHEN_EXCEPTION);            }        };        boolean commitOffsetEnable = false;        long commitOffsetValue = 0L;        if (MessageModel.CLUSTERING == this.defaultMQPushConsumer.getMessageModel()) {            commitOffsetValue = this.offsetStore.readOffset(pullRequest.getMessageQueue(), ReadOffsetType.READ_FROM_MEMORY);            if (commitOffsetValue > 0) {                commitOffsetEnable = true;            }        }        String subExpression = null;        boolean classFilter = false;        SubscriptionData sd = this.rebalanceImpl.getSubscriptionInner().get(pullRequest.getMessageQueue().getTopic());        if (sd != null) {            if (this.defaultMQPushConsumer.isPostSubscriptionWhenPull() && !sd.isClassFilterMode()) {                subExpression = sd.getSubString();            }            classFilter = sd.isClassFilterMode();        }        int sysFlag = PullSysFlag.buildSysFlag(            commitOffsetEnable, // commitOffset            true, // suspend            subExpression != null, // subscription            classFilter // class filter        );        try {            // 下面我們看繼續(xù)跟進(jìn)這個(gè)方法，這個(gè)方法已經(jīng)就是客戶端如何拉取消息            this.pullAPIWrapper.pullKernelImpl(                pullRequest.getMessageQueue(),                subExpression,                subscriptionData.getExpressionType(),                subscriptionData.getSubVersion(),                pullRequest.getNextOffset(),                this.defaultMQPushConsumer.getPullBatchSize(),                sysFlag,                commitOffsetValue,                BROKER_SUSPEND_MAX_TIME_MILLIS,                CONSUMER_TIMEOUT_MILLIS_WHEN_SUSPEND,                // 消息的通信方式為異步                CommunicationMode.ASYNC,                pullCallback            );        } catch (Exception e) {            log.error("pullKernelImpl exception", e);            this.executePullRequestLater(pullRequest, PULL_TIME_DELAY_MILLS_WHEN_EXCEPTION);        }    }

發(fā)送遠(yuǎn)程請求拉取消息

在MQClientAPIImpl.pullMessage方法中，根據(jù)入?yún)ommunicationMode的值分為異步拉取和同步拉取方式兩種。

無論是異步方式拉取還是同步方式拉取，在發(fā)送拉取請求之前都會構(gòu)造一個(gè)ResponseFuture對象，以請求消息的序列號為key值，存入NettyRemotingAbstract.responseTable:ConcurrentHashMap, ResponseFuture>變量中，對該變量有幾種情況會處理：

1. 發(fā)送失敗后直接刪掉responseTable變量中的相應(yīng)記錄；
2. 收到響應(yīng)消息之后，會以響應(yīng)消息中的序列號（由服務(wù)端根據(jù)請求消息的序列號原樣返回）從responseTable中查找ResponseFuture對象，并設(shè)置該對象的responseCommand變量。若是同步發(fā)送會喚醒等待響應(yīng)的ResponseFuture.waitResponse方法；若是異步發(fā)送會調(diào)用ResponseFuture.executeInvokeCallback()方法完成回調(diào)邏輯處理；
3. 在NettyRemotingClient.start()啟動(dòng)時(shí)，也會初始化定時(shí)任務(wù)，該定時(shí)任務(wù)每隔1秒定期掃描responseTable列表，遍歷該列表中的ResponseFuture對象，檢查等待響應(yīng)是否超時(shí)，若超時(shí)，則調(diào)用ResponseFuture. executeInvokeCallback()方法，并將該對象從responseTable列表中刪除；

public PullResult pullMessage(        final String addr,        final PullMessageRequestHeader requestHeader,        final long timeoutMillis,        final CommunicationMode communicationMode,        final PullCallback pullCallback    ) throws RemotingException, MQBrokerException, InterruptedException {        RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.PULL_MESSAGE, requestHeader);        switch (communicationMode) {            case ONEWAY:                assert false;                return null;            case ASYNC:                this.pullMessageAsync(addr, request, timeoutMillis, pullCallback);                return null;            case SYNC:                return this.pullMessageSync(addr, request, timeoutMillis);            default:                assert false;                break;        }        return null;    }

同步拉取

對于同步發(fā)送方式，調(diào)用MQClientAPIImpl.pullMessageSync(String addr, RemotingCommand request, long timeoutMillis)方法,大致步驟如下：

1. 調(diào)用RemotingClient.invokeSync(String addr, RemotingCommand request, long timeoutMillis)方法：
   • 獲取Broker地址的Channel信息。根據(jù)broker地址從RemotingClient.channelTables:ConcurrentHashMap, ChannelWrapper>變量中獲取ChannelWrapper對象并返回該對象的Channel變量；若沒有ChannelWrapper對象則與broker地址建立新的連接并將連接信息存入channelTables變量中，便于下次使用；
   • 若NettyRemotingClient.rpcHook:RPCHook變量不為空（該變量在應(yīng)用層初始化DefaultMQPushConsumer或者DefaultMQPullConsumer對象傳入該值），則調(diào)用RPCHook.doBeforeRequest(String remoteAddr, RemotingCommand request)方法；
   • 調(diào)用NettyRemotingAbstract.invokeSyncImpl(Channel channel, RemotingCommand request, long timeoutMillis)方法，該方法的邏輯如下：
     • A）使用請求的序列號（opaue）、超時(shí)時(shí)間初始化ResponseFuture對象；并將該ResponseFuture對象存入NettyRemotingAbstract.responseTable: ConcurrentHashMap變量中；
     • B）調(diào)用Channel.writeAndFlush(Object msg)方法將請求對象RemotingCommand發(fā)送給Broker；然后調(diào)用addListener(GenericFutureListener> listener)方法添加內(nèi)部匿名類：該內(nèi)部匿名類實(shí)現(xiàn)了ChannelFutureListener接口的operationComplete方法，在發(fā)送完成之后回調(diào)該監(jiān)聽類的operationComplete方法，在該方法中，首先調(diào)用ChannelFuture. isSuccess()方法檢查是否發(fā)送成功，若成功則置ResponseFuture對象的sendRequestOK等于true并退出此回調(diào)方法等待響應(yīng)結(jié)果；若不成功則置ResponseFuture對象的sendRequestOK等于false，然后從NettyRemotingAbstract.responseTable中刪除此請求序列號（opaue）的記錄，置ResponseFuture對象的responseCommand等于null，并喚醒ResponseFuture.waitResponse(long timeoutMillis)方法的等待；
     • C）調(diào)用ResponseFuture.waitResponse(long timeoutMillis)方法等待響應(yīng)結(jié)果；在發(fā)送失敗或者收到響應(yīng)消息（詳見5.10.3小節(jié)）或者超時(shí)的情況下會喚醒該方法返回ResponseFuture.responseCommand變量值；
     • D）若上一步返回的responseCommand值為null，則拋出異常:若ResponseFuture.sendRequestOK為true，則拋出RemotingTimeoutException異常，否則拋出RemotingSendRequestException異常；
     • E）若上一步返回的responseCommand值不為null，則返回responseCommand變量值；
   • 若NettyRemotingClient.rpcHook: RPCHook變量不為空，則調(diào)用RPCHook.doAfterResponse(String remoteAddr, RemotingCommand request)方法；

• 以上一步的返回值RemotingCommand對象為參數(shù)調(diào)用MQClientAPIImpl. processPullResponse (RemotingCommand response)方法將返回對象解析并封裝成PullResultExt對象然后返回給調(diào)用者，響應(yīng)消息的結(jié)果狀態(tài)轉(zhuǎn)換如下：
  • 若RemotingCommand對象的Code等于SUCCESS，則PullResultExt.pullStatus=FOUND；
  • 若RemotingCommand對象的Code等于PULL_NOT_FOUND，則PullResultExt.pullStatus= NO_NEW_MSG；
  • 若RemotingCommand對象的Code等于PULL_RETRY_IMMEDIATELY，則PullResu