// org.apache.rocketmq.client.consumer.MQConsumer
public interface MQConsumer extends MQAdmin {
    // 如果消費(fèi)失敗，消息將被發(fā)送回代理，并延遲消耗一些時(shí)間
    void sendMessageBack(final MessageExt msg/*消息*/, final int delayLevel/*延遲級(jí)別*/, final String brokerName);  
    // 根據(jù)topic從使用者緩存中獲取消息隊(duì)列
    Set<MessageQueue> fetchSubscribeMessageQueues(final String topic) throws MQClientException;
}
// org.apache.rocketmq.client.consumer.MQPushConsumer
public interface MQPushConsumer extends MQConsumer {
    // 啟動(dòng)Consumer
    void start() throws MQClientException;
    // 關(guān)閉Consumer
    void shutdown();
    // 注冊(cè)并發(fā)消息Listener
    void registerMessageListener(final MessageListenerConcurrently messageListener);
    // 注冊(cè)順序消息Listener,將會(huì)有序地接收消息。一個(gè)隊(duì)列一個(gè)線程
    void registerMessageListener(final MessageListenerOrderly messageListener);
    // 訂閱Topic
    void subscribe(final String topic, final String subExpression) throws MQClientException;
    // 退訂topic
    void unsubscribe(final String topic);
}

// org.apache.rocketmq.client.consumer.DefaultMQPushConsumer
public class DefaultMQPushConsumer extends ClientConfig implements MQPushConsumer {
    // DefaultMQPushConsumer的默認(rèn)實(shí)現(xiàn)，DefaultMQPushConsumer中大部分功能都是對(duì)它的代理
    protected final transient DefaultMQPushConsumerImpl defaultMQPushConsumerImpl;
    // 相同角色的消費(fèi)者需要具有完全相同的subscriptions和consumerGroup才能正確實(shí)現(xiàn)負(fù)載平衡，它需要全局唯一
    private String consumerGroup;
    // 消息模型定義了如何將消息傳遞到每個(gè)消費(fèi)者客戶端的方式，默認(rèn)是集群模式
    private MessageModel messageModel = MessageModel.CLUSTERING;
  	// 第一次消費(fèi)時(shí)指定的消費(fèi)策略，默認(rèn)是CONSUME_FROM_LAST_OFFSET
    private ConsumeFromWhere consumeFromWhere = ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET;
  	// 隊(duì)列分配算法，指定如何將消息隊(duì)列分配給每個(gè)使用者客戶端。
    private AllocateMessageQueueStrategy allocateMessageQueueStrategy;
    // 訂閱關(guān)系
    private Map<String /* topic */, String /* sub expression */> subscription = new HashMap<String, String>();
    // 消息監(jiān)聽器
    private MessageListener messageListener;
    // 消息消費(fèi)進(jìn)度存儲(chǔ)器
    private OffsetStore offsetStore;
  	// 最小消費(fèi)線程數(shù)
    private int consumeThreadMin = 20;
    // 最大消費(fèi)線程數(shù)
    private int consumeThreadMax = 20;
    // 推送模式下拉去消息的間隔時(shí)間，默認(rèn)一次拉取消息完成后立刻繼續(xù)拉取
    private long pullInterval = 0;
    // 批量消費(fèi)數(shù)量
    private int consumeMessageBatchMaxSize = 1;
    // 批量拉取的數(shù)量
    private int pullBatchSize = 32;
  	// 每次拉取時(shí)是否更新訂閱關(guān)系,默認(rèn)是false
    private boolean postSubscriptionWhenPull = false;
    // 消息最大重試次數(shù)，如果消息消費(fèi)最大次數(shù)超過(guò)maxReconsumeTimes還未成功，則消息將被轉(zhuǎn)移到一個(gè)失敗隊(duì)列
    private int maxReconsumeTimes = -1;
    //延遲將該隊(duì)列的消息提交到消費(fèi)者線程的等待時(shí)間，默認(rèn)延遲1s
    private long suspendCurrentQueueTimeMillis = 1000;
    // 消息阻塞消費(fèi)線程的最大超時(shí)時(shí)間，默認(rèn)15分鐘
    private long consumeTimeout = 15;
    // 關(guān)閉使用者時(shí)等待消息的最長(zhǎng)時(shí)間，0表示沒有等待。
    private long awaitTerminationMillisWhenShutdown = 0;
}

public static void main(String[] args) throws MQClientException {
    // 1. 傳入CONSUMER_GROUP，創(chuàng)建DefaultMQPushConsumer
    DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(CONSUMER_GROUP);
  	// 2. 設(shè)置namesrvAddr
  	consumer.setNamesrvAddr("127.0.0.1:9876");
    // 3. 訂閱Topic 
    consumer.subscribe(TOPIC, "*");
    // 4.注冊(cè)消息Listener
    consumer.registerMessageListener((MessageListenerConcurrently) (msg, context) -> {
        System.out.printf("%s Receive New Messages: %s %n", Thread.currentThread().getName(), msg);
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    });
    // 5.啟動(dòng)Consumer
    consumer.start();
}

// org.apache.rocketmq.client.consumer.DefaultMQPushConsumer#start
public void start() throws MQClientException {
    // consumerGroup封裝namespace
    setConsumerGroup(NamespaceUtil.wrapNamespace(this.getNamespace(), this.consumerGroup));
    // DefaultMQPushConsumerImpl啟動(dòng)
    this.defaultMQPushConsumerImpl.start();
    // 消息軌跡跟蹤服務(wù)，默認(rèn)null
    if (null != traceDispatcher) {
        try {
            traceDispatcher.start(this.getNamesrvAddr(), this.getAccessChannel());
        } catch (MQClientException e) {
            log.warn("trace dispatcher start failed ", e);
        }
    }
}

// org.apache.rocketmq.client.impl.consumer.DefaultMQPushConsumerImpl#start
public synchronized void start() throws MQClientException {
  //...
  // 狀態(tài)先設(shè)置為啟動(dòng)失敗
  this.serviceState = ServiceState.START_FAILED;
  // 校驗(yàn)配置，ConsumerGroup校驗(yàn)，
  this.checkConfig();
  // 訂閱關(guān)系copy到RebalanceImpl中
  this.copySubscription();
  // 如果是集群模式，消費(fèi)者名稱如果是DEFAULT，則會(huì)改成：PID#時(shí)間戳
  if (this.defaultMQPushConsumer.getMessageModel() == MessageModel.CLUSTERING) {
      this.defaultMQPushConsumer.changeInstanceNameToPID();
  }
  //...
}

// org.apache.rocketmq.client.impl.consumer.DefaultMQPushConsumerImpl#start
// 生成一個(gè)MQClientInstance
this.mQClientFactory = MQClientManager.getInstance().getOrCreateMQClientInstance(this.defaultMQPushConsumer, this.rpcHook);
// 設(shè)置消費(fèi)者組
this.rebalanceImpl.setConsumerGroup(this.defaultMQPushConsumer.getConsumerGroup());
// 消息消費(fèi)模式
this.rebalanceImpl.setMessageModel(this.defaultMQPushConsumer.getMessageModel());
// 設(shè)置消息消費(fèi)模式
this.rebalanceImpl.setAllocateMessageQueueStrategy(this.defaultMQPushConsumer.getAllocateMessageQueueStrategy());
// 設(shè)置MQClientInstance
this.rebalanceImpl.setmQClientFactory(this.mQClientFactory);
// 構(gòu)建拉消息包裝器
this.pullAPIWrapper = new PullAPIWrapper(
    mQClientFactory,
    this.defaultMQPushConsumer.getConsumerGroup(), isUnitMode());
this.pullAPIWrapper.registerFilterMessageHook(filterMessageHookList);

// org.apache.rocketmq.client.impl.consumer.DefaultMQPushConsumerImpl#start
if (this.defaultMQPushConsumer.getOffsetStore() != null) {
    this.offsetStore = this.defaultMQPushConsumer.getOffsetStore();
} else {
    switch (this.defaultMQPushConsumer.getMessageModel()) {
        case BROADCASTING:
            // 如果是廣播模式，則使用LocalFileOffsetStore存儲(chǔ)偏移量
            this.offsetStore = new LocalFileOffsetStore(this.mQClientFactory, this.defaultMQPushConsumer.getConsumerGroup());
            break;
        case CLUSTERING:
            // 如果是集群模式，則使用RemoteBrokerOffsetStore存儲(chǔ)偏移量
            this.offsetStore = new RemoteBrokerOffsetStore(this.mQClientFactory, this.defaultMQPushConsumer.getConsumerGroup());
            break;
        default:
            break;
    }
    this.defaultMQPushConsumer.setOffsetStore(this.offsetStore);
}
// 如果是廣播模式，則從本地文件load偏移量，如果是集群模式則是一個(gè)空實(shí)現(xiàn)
this.offsetStore.load();

// org.apache.rocketmq.client.impl.consumer.DefaultMQPushConsumerImpl#start
// 如果是順序消費(fèi)
if (this.getMessageListenerInner() instanceof MessageListenerOrderly) {
    this.consumeOrderly = true;
    this.consumeMessageService =
        new ConsumeMessageOrderlyService(this, (MessageListenerOrderly) this.getMessageListenerInner());
// 如果是并發(fā)消費(fèi)
} else if (this.getMessageListenerInner() instanceof MessageListenerConcurrently) {
    this.consumeOrderly = false;
    this.consumeMessageService =
        new ConsumeMessageConcurrentlyService(this, (MessageListenerConcurrently) this.getMessageListenerInner());
}
this.consumeMessageService.start();
// 將自身注冊(cè)到MQClientInstance
boolean registerOK = mQClientFactory.registerConsumer(this.defaultMQPushConsumer.getConsumerGroup(), this);
// ...
mQClientFactory.start();

// org.apache.rocketmq.client.impl.consumer.DefaultMQPushConsumerImpl#start
// 向Namesrv拉取并更新當(dāng)前消費(fèi)者訂閱topic路由信息
this.updateTopicSubscribeInfoWhenSubscriptionChanged();
// 隨機(jī)選擇一個(gè)Broker，發(fā)送檢查客戶端tag配置的請(qǐng)求，主要是檢測(cè)Broker是否支持SQL92類型的tag過(guò)濾以及SQL92的tag語(yǔ)法是否正確
this.mQClientFactory.checkClientInBroker();
// 給所有Broker發(fā)送心跳
this.mQClientFactory.sendHeartbeatToAllBrokerWithLock();
// 喚醒負(fù)載均衡服務(wù)rebalanceService，并進(jìn)行rebalance
this.mQClientFactory.rebalanceImmediately();