原文地址：http://adamswanglin.com/rocketmq/rocketmq-consumer/

關於rocketmq-client包

RocketMQ將producer，consumer和admin相關程式碼都放到rocketmq-client jar包裡；RocketMQ的採用客戶端拉的方式消費訊息（PUSH也是通過客戶端拉來實現的），拉取的時候要考慮負載均衡（rebalance），考慮訊息至少消費一次（offset管理）；等等這些導致了consumer的複雜度是client裡面最高的。

Consumer三個重要部分

負載均衡

負載均衡狀態

假設上面是執行一段時間的狀態，C_ORD消費組有兩個節點node1和node2，訂閱了topic：TP_PAY，Broker為TP_PAY建立了四個Queue，node1消費Queue1和Queue2，node2消費Queue3和Queue4。

Rebalance過程

客戶端每20s會啟動rebalance，節點rebalance過程是：

需要說明的是以上過程，各個節點都是分別進行的。

節點變動

實際中節點會上下線，節點數量會發生變動，Topic配置的queue數量也可能變動（Broker配置變更或者上下線）。這時候Rebalance過程中各個消費節點消費的queue就會發生變動。

考慮上面平衡狀態下，C_ORD消費組中新增了節點node3：

node3啟動的時候會Rebalance，發現自己應該訂閱Queue4；因為node3從沒有訂閱到訂閱了Queue4，訂閱內容有變，所以會node3立即傳送心跳給Broker並且其中subVersion是當前時間戳；Broker發現心跳中的時間戳有更新會立即傳送NOTIFY_CONSUMER_IDS_CHANGED的指令給node1和node2；node1和node2收到指令會立即Rebalance。只要同一個消費組的訂閱資訊一致，分配演算法一致，最終queue會被同一消費組的節點平均分配。最終變化如下：

負載均衡演算法有：

AllocateMessageQueueAveragely

AllocateMessageQueueAveragelyByCircle

AllocateMessageQueueConsistentHash

AllocateMessageQueueByMachineRoom

AllocateMessageQueueByConfig

預設是AllocateMessageQueueAveragely；例如5個queue分給3個節點，5/3=1，則平均消費1個；5%3=2，頭兩個節點再額外加一個；最後的結果是{1,2} {3,4} {5}。

訊息拉取

PULL OR PUSH

consumer分pull consumer

• pull：應用自己拉取，訊息延時較大。
• push：傳送非同步拉取訊息的請求給Broker；如果當時有未消費的訊息Broker立即返回未消費訊息，如果沒有未消費的訊息，Broker在有新訊息的時候返回新訊息。push型別的consumer實現中的非同步拉取訊息的請求實際也是客戶端pull訊息；使用這種實現方式減輕了Broker的負擔（Broker通過請求就知道新訊息發給誰），但也增加了客戶端的複雜度（負載均衡在客戶端實現）。

我們使用的都是pushConsumer，一下只考慮pushConsumer的啟動。

BROADCASTING OR CLUSTERING

consumer的消費模型分為BROADCASTING（廣播模式）和CLUSTERING（叢集模式）。

• 廣播模式：對於某一條訊息，同一消費組裡每個節點都會收到。

• 叢集模式：對於某一條訊息，同一消費組裡只會有一個節點收到。

廣播模式和叢集模式只針對單個消費組；不同消費組之間的訊息訂閱是互不影響的。

Offset管理

為了確定訊息是否到達，現有訊息佇列實現裡都有ack機制。

例如在ActiveMQ裡，一條訊息從producer端發出之後，一旦被consumer消費，consumer會返回ACK，broker端會刪除這條已消費的訊息。這樣每一條訊息消費都要傳送一個ACK訊息，Broker端也要根據ACK做相應操作。

RocketMQ用Offset機制來實現ACK，它類似一種批量的ACK：

• 在Broker端，訊息的Offset是遞增的；

• Client端拉取的時候也是按順序拉取的，比如第一次拉取offset 0開始的訊息，拉取了10條，第二次就從上次最後一個節點offset+1的位置拉取；

• Client消費一批訊息後將消費完成的Offset傳送給Broker。

RocketMQ這樣做之後提升了效率：Offset更新頻率相比單條更新小，Broker端只用儲存某個消費組對某個Queue的消費進度而不用在每個訊息上存某個消費組是否消費了該訊息。

但同時也帶來了一個問題：更大機率的重複消費。消費組消費了offset =2到offset=10的訊息，但是offset=1的訊息消費的比較慢；如果更新offset=10可能會導致offset=1的訊息未成功消費Broker卻認為成功，所以RocketMQ的做法是消費端更新的Offset都是未消費訊息的最小offset；如果這時候消費端down機，別的消費組消費的時候會從offset=1的訊息開始拉取消費，這樣offset=2和offset=10的訊息就會重複消費。所以RocketMQ不保證訊息不重複，當然這只是造成訊息重複消費的一個原因。

Consumer啟動

幾個重要的類關係

PullMessageService

啟動拉取訊息執行緒類

RebalanceService

啟動定時任務Rebalance類

MQClientInstance

一般一個應用一個

DefaultMQPushConsumerImpl

一個consumerGroup一個，和MQClientInstance是多對一關係

RebalanceImpl

和pull/push模式有關，所以具體的rebalance實現掛在DefaultMQPushConsumerImpl下面。

啟動詳細流程

啟動DefaultMQPushConsumerImpl

獲取/建立MQClientInstance

MQClientInstance和clientId一一對應，一般一個應用只會有MQClientInstance，規則是{ip}:{pid}。

建立MQClientAPI

封裝了遠端呼叫Broker和NameServer的API。包括推送訊息，拉取訊息，在Broker上建立消費組等。

建立PullMessageService

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

public class PullMessageService extends ServiceThread {
    @Override
    public void run() {
        while (!this.isStopped()) {
            try {
                PullRequest pullRequest = this.pullRequestQueue.take();
                if (pullRequest != null) {
                    this.pullMessage(pullRequest);
                }
            } catch (InterruptedException e) {
            } catch (Exception e) {
                log.error("Pull Message Service Run Method exception", e);
            }
        }
    }
}

其中ServiceThread是對執行緒的封裝，RocketMQ中很多XXService類都繼承自ServiceThread。

ServiceTask實現程式開啟，停止，等待特定時長執行，中途任意時間喚醒等。

擷取部分程式碼如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36

public abstract class ServiceThread implements Runnable {
	protected final Thread thread;
  	//和juc CountDownLatch 區別 增加了 reset方法
    protected final CountDownLatch2 waitPoint = new CountDownLatch2(1);
    protected volatile AtomicBoolean hasNotified = new AtomicBoolean(false);

    public void start() {
        this.thread.start();
    }
    //執行緒等待程式碼
  	protected void waitForRunning(long interval) {
        if (hasNotified.compareAndSet(true, false)) {
            this.onWaitEnd();
            return;
        }

        //entry to wait
        waitPoint.reset();

        try {
            waitPoint.await(interval, TimeUnit.MILLISECONDS);
        } catch (InterruptedException e) {
            log.error("Interrupted", e);
        } finally {
            hasNotified.set(false);
            this.onWaitEnd();
        }
    }
  //執行緒中途喚醒程式碼
  	public void wakeup() {
        if (hasNotified.compareAndSet(false, true)) {
            waitPoint.countDown(); // notify
        }
    }
}

能看到PullMessageService的作用是啟動一個執行緒，不停的從queue里拉取請求並執行pullMessage方法。

建立RebalanceService

1
2
3
4
5
6
7
8
9

public class RebalanceService extends ServiceThread {
    @Override
    public void run() {
        while (!this.isStopped()) {
            this.waitForRunning(waitInterval);
            this.mqClientFactory.doRebalance();
        }
    }
}

RebalanceService的作用就是啟動一個執行緒，定時呼叫doRebalance方法。

建立PullApiWrapper

拉取訊息請求和響應的wrapper；主要作用：請求的時候封裝RPC請求體，響應的時候二次過濾tag。

說二次過濾是因為Broker會過濾一次tag，但是為了效率broker過濾tag是按存放的tag的hashCode做比較的，不保證tag字串一致。

PullApiWrapper做二次過濾保證獲取的的tag和訂閱的tag字串匹配。

啟動ConsumeMessageConcurrentlyService

訊息消費處理類，這裡是併發處理訊息；對應的還有ConsumeMessageOrderlyService，順序處理訊息類。

除了處理下訊息外，還負責啟動執行緒定時清除消費開始15分鐘還未處理完成的訊息（傳送回Broker重試）。

啟動MQClientInstance

啟動netty客戶端；啟動拉取訊息服務；啟動rebalance服務；啟動定時任務：定時向Broker傳送心跳，定時拉取路由資訊，定時傳送offset到Broker，定時調整消費執行緒池大小。

registerConsumer

DefaultMQPushConsumerImpl是一個consumerGroup一個例項，MQClientInstance一個應用一個例項；兩者一對多，註冊指DefaultMQPushConsumerImpl放入MQClientInstance中的ConcurrentMap<String/* group */, MQConsumerInner>中。

updateTopicRouteInfoFromNameServer

立即更新一次訂閱的topic的路由資訊。

checkClientInBroker

隨機選擇一個Broker，傳送檢查客戶端配置配置的請求。

sendHeartbeatToAllBrokerWithLock

立即向所有相關Broker（訂閱的topic的路由到的Broker）的master節點發送心跳。

rebalanceImmediately

立即執行一次rebalance。

PushConsumerRebalance

Consumer拉取訊息

拉取訊息的流程示意

圖中小人代表有特定執行緒處理任務；黃色箭頭代表PullRequest的流向。

RebalanceService初始化PullRequest

client啟動

RebalanceService確定consumer拉取的queue。

• 為需要拉取的queue生成一個ProcessQueue用來儲存正在/等待處理的資訊，放入processQueueTable中。
• 為需要拉取的queue生成一個PullRequest，放入PullRequestQueue中；其中，拉取訊息的位置從nextOffset從Broker遠端拉取。

定時Rebalance

• 如果發現有新訂閱的queue，也會為每個新增訂閱的queue生成一個PullRequest，放入PullRequestQueue中；其中，拉取訊息的位置從nextOffset從Broker遠端拉取。
• 如果發現有queue已經不訂閱了，更新offset到Broker，將ProcessQueue設為dropped並從processQueueTable中移除。

PullMessageService取出PullRequest

PullMessageService啟動一個執行緒不停的從PullRequestQueue裡取出PullRequest。如果取出的PullRequest是已失效的（ProcessQueue是否dropped），丟棄；如果未失效，執行下面步驟。

PullMessageService傳送非同步請求

PullMessageService取出PullRequest後，根據其中的queue定位Broker，併發送非同步拉取請求。同時將PullRequest封裝在PullCallback裡，PullCallback封裝在ResponseFuture裡；並以自增的請求id為鍵，ResponseFuture為值放入ResponseTable中。

Broker傳送非同步響應

Broker收到請求，如果offset之後有新的訊息會立即傳送非同步響應；否則等待直到producer有新的訊息傳送後返回或者超時。

如果通訊異常或者Broker超時未返回響應，nettyClient會定時清理超時的請求，釋放PullRequest回到PullRequestQueue。

NettyClient處理響應

根據響應id從ResponseTable中取出ResponseFuture；從響應裡取出最新的offset和批量拉取到的訊息。

用最新的offset更新ResponseFuture裡的PullRequest並推送給PullRequestQueue裡以進行下一次拉取。

批量拉取到的訊息分批推給consumeExecutor執行緒處理。

拉取訊息的詳細流程

幾點說明

• 拉取到的訊息分批用consumerExecutor執行緒池執行，如果執行緒池滿了5s後重試。

• 每批訊息裡消費失敗的訊息會重發給broker的重試佇列，重發也失敗的訊息5s後用consumerExecutor重新消費。

• 每批消費完成後更新offset到Broker。

• 非同步收到訊息後，訊息會分queue放到ProcessQueue中，ProcessQueue裡的msgTreeMap：TreeMap<Long/offset/, MessageExt>存放訊息；收到的訊息處理成功或者處理失敗已發回重試會從treemap裡移除。幾點需要注意：

  1.每次更新offset到Broker都是從treemap裡取第一條（最小offset），某條訊息消費超時會導致Broker的offset無法更新；當某條消費超過15分鐘還未消費完成，會發回Broker嘗試重試。

  2.treemap裡堆積的訊息超過1000（可配）條或者最小最大offset相差超過2000（可配）會觸發控流，延遲拉取訊息。

• 關於訊息重複：

  RocketMQ不保證訊息不重複，如果你的業務需要保證嚴格的不重複訊息，需要你自己在業務端去重。

• 關於訊息消費順序：

  RocketMQ有嚴格順序消費的實現。但是有序消費會影響訊息並行處理效率，消費端吞吐量下降；而且單條訊息阻塞會阻塞這個消費端；所以我們並沒有使用訊息順序消費。

Consumer關閉

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

//DefaultMQPushConsumerImpl - shutdown
public synchronized void shutdown() {
  switch (this.serviceState) {
    case CREATE_JUST:
      break;
    case RUNNING:
      //關閉訊息消費執行緒池
      this.consumeMessageService.shutdown();
      //消費進度同步到Broker
      this.persistConsumerOffset();
//在Broker裡取消註冊
      this.mQClientFactory.unregisterConsumer(this.defaultMQPushConsumer.getConsumerGroup());
      this.mQClientFactory.shutdown();
      log.info("the consumer [{}] shutdown OK", this.defaultMQPushConsumer.getConsumerGroup());
      //關閉ProcessQueue
      this.rebalanceImpl.destroy();
      this.serviceState = ServiceState.SHUTDOWN_ALREADY;
      break;
    case SHUTDOWN_ALREADY:
      break;
    default:
      break;
  }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30

synchronized (this) {
  switch (this.serviceState) {
    case CREATE_JUST:
      break;
    case RUNNING:
      //關閉生產者
      this.defaultMQProducer.getDefaultMQProducerImpl().shutdown(false);
      this.serviceState = ServiceState.SHUTDOWN_ALREADY;
      //關閉拉取訊息執行緒
      this.pullMessageService.shutdown(true);
      //關閉定時任務--heartbeat updateRouteInfo persistAllConsumerOffset等
      this.scheduledExecutorService.shutdown();
      //關閉netty服務
      this.mQClientAPIImpl.shutdown();
      //關閉rebalance執行緒
      this.rebalanceService.shutdown();

      if (this.datagramSocket != null) {
        this.datagramSocket.close();
        this.datagramSocket = null;
      }
      MQClientManager.getInstance().removeClientFactory(this.clientId);
      log.info("the client factory [{}] shutdown OK", this.clientId);
      break;
    case SHUTDOWN_ALREADY:
      break;
    default:
      break;
  }
}

關於rocketmq-client包

RocketMQ將producer，consumer和admin相關程式碼都放到rocketmq-client jar包裡；RocketMQ的採用客戶端拉的方式消費訊息（PUSH也是通過客戶端拉來實現的），拉取的時候要考慮負載均衡（rebalance），考慮訊息至少消費一次（offset管理）；等等這些導致了consumer的複雜度是client裡面最高的。

Consumer三個重要部分

相關推薦