概览
消息消费包含两种推模式和拉模式，拉模式官方已经不推荐使用，那我们今天就介绍推模式。DefaultMQPushConsumer是推模式消费的对外API，消费的功能主要由成员变量DefaultMQPushConsumerImpl提供。
DefaultMQPushConsumerImpl是消息消费的主要实现类。消费者启动流程关键步骤如下：
this.copySubscription();订阅重试主题消息。从这里可以看出，RocketMQ消息重试是以消费组为单位的，而不是主题，消息重试主题名为"%RETRY%"+消费组名，消费者在启动的时候会自动订阅该主题，参与该主题的消息队列负载。
创建MQClientInstance实例，封装了 RocketMQ 网络处理 API，是消息生产者、消息消费者与 NameServer、Broker 打交道的网络通道。
创建PullAPIWrapper对象，拉取Broker消息的API操作从包装器。
OffsetStore对象，初始化消息消费进度。如果消息消费是集群模式，那么消息进度保存在Broker上，如果是广播模式，那么消息消费进度存储在消费端。
从磁盘加载消费进度文件。
创建ConsumeMessageService对象并开启消费消息服务。根据是否是顺序消费，创建消费端消费线程服务。consumeMessageService主要负责消息消费，内部维护一个线程池。
启动MQClientInstance服务，MQ客户端的管理者。
// 启动消费者
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();
// 构建主题订阅信息SubscriptionData并加入到RebalanceImpl的订阅信息中
this.copySubscription();
// 消费者集群模式
if (this.defaultMQPushConsumer.getMessageModel() == MessageModel.CLUSTERING) {
this.defaultMQPushConsumer.changeInstanceNameToPID();
}
// MQClient工厂，初始化MQClientInstance、RebalancePushImpl
this.mQClientFactory = MQClientManager.getInstance().getOrCreateMQClientInstance(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);
// 初始化消息消费进度。如果消息消费是集群模式，那么消息进度保存在Broker上，如果是广播模式，那么消息消费进度存储在消费端。
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();
// 根据是否是顺序消费，创建消费端消费线程服务。consumeMessageService主要负责消息消费，内部维护一个线程池
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();
//向MQClientInstance注册消费者，并启动MQClientInstance，在一个JVM一个实例，所有消费者、生产者持有同一个MQClientInstance，MQClientInstance只会启动一次。
boolean registerOK = mQClientFactory.registerConsumer(this.defaultMQPushConsumer.getConsumerGroup(), this);
if (!registerOK) {
this.serviceState = ServiceState.CREATE_JUST;
this.consumeMessageService.shutdown(defaultMQPushConsumer.getAwaitTerminationMillisWhenShutdown());
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);
}
// 启动MQClientInstance实例
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;
}
// 根据订阅信息的topic去更新路由信息
this.updateTopicSubscribeInfoWhenSubscriptionChanged();
this.mQClientFactory.checkClientInBroker();
// Consumer发送心跳到所有broker上
this.mQClientFactory.sendHeartbeatToAllBrokerWithLock();
// 立即消费者和消费队列负载均衡
this.mQClientFactory.rebalanceImmediately();
}
MQClientInstance封装了 RocketMQ 网络处理 API，是消息生产者、消息消费者与 NameServer、Broker 打交道的网络通道。同一个 JVM 中的不同消费者和不同生产者在启动时获取到的 MQClientInstance 实例都是同一个。
MQClientInstance的启动关键步骤：
- MQClientAPIImpl开启服务，生产者和消费者客户端消息发送和消费的处理的API
- 开启定时任务，主要和延时消息相关，延时消息后面会详细介绍。
- 开启从Broker拉取消息服务，供消费端消息消费。
- 开启消费者和消费队列的消息消费的负载均衡服务。
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;
}
}
}
总结
DefaultMQPushConsumerImpl主要用来是客户端进行消息消费的，这里创建了ConsumeMessageService消息消费服务、消息进度保存对象OffsetStore，消息消费的监听器对象MessageListener。
MQClientInstance主要是用来从Broker拉取消息、消费者和消息队列负载均衡的，开启了消息拉取服务PullMessageService、负载均衡服务RebalanceService，为消息消费做准备的。
DefaultMQPushConsumerImpl和MQClientInstance都是部署在客户端的，包括从Broker拉取消息，消息队列的负载均衡都是再客户端完成的。
下面的章节将详细介绍消费者和消息队列的负载均衡、消费组消息拉取与消费。
