第十六章-消费者-PUSH方式（一）

16.1 准备阶段

先从一段官方示例代码开始

public class Consumer {public static void main(String[] args) throws InterruptedException, MQClientException {// 初始化consumer，并设置consumer group nameDefaultMQPushConsumer consumer = new DefaultMQPushConsumer("please_rename_unique_group_name");// 设置NameServer地址 consumer.setNamesrvAddr("localhost:9876");//订阅一个或多个topic，并指定tag过滤条件，这里指定*表示接收所有tag的消息consumer.subscribe("TopicTest", "*");//注册回调接口来处理从Broker中收到的消息consumer.registerMessageListener(new MessageListenerConcurrently() {@Overridepublic ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {System.out.printf("%s Receive New Messages: %s %n", Thread.currentThread().getName(), msgs);// 返回消息消费状态，ConsumeConcurrentlyStatus.CONSUME_SUCCESS为消费成功return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;}});// 启动Consumerconsumer.start();System.out.printf("Consumer Started.%n");}
}

更多的介绍，可以直接看官网，咱们这直接上源码，那就从 consumer.start() 这一行代码开始。

DefaultMQPushConsumer.start

public void start() throws MQClientException {// 通过命名空间处理一下消费者组setConsumerGroup(NamespaceUtil.wrapNamespace(this.getNamespace(), this.consumerGroup));// push消费的实现类启动this.defaultMQPushConsumerImpl.start();// 访问日志追踪if (null != traceDispatcher) {try {traceDispatcher.start(this.getNamesrvAddr(), this.getAccessChannel());} catch (MQClientException e) {log.warn("trace dispatcher start failed ", e);}}
}

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();// 复制订阅信息，这里所做的事就是将订阅信息从 DefaultMQPushConsumerImpl 类的map 复制到 RebalanceImpl 类的map（rebalanceImpl.getSubscriptionInner()）中，因为最终做消息处理的是 RebalanceImpl去与服务端打交道this.copySubscription();// 集群模式，实例名改成对应的进程Idif (this.defaultMQPushConsumer.getMessageModel() == MessageModel.CLUSTERING) {this.defaultMQPushConsumer.changeInstanceNameToPID();}// 获取或创建MQClientInstance实例，这个类很重要，同时涵盖了生产者和消费者的操作，具体使用时会讲this.mQClientFactory = MQClientManager.getInstance().getAndCreateMQClientInstance(this.defaultMQPushConsumer, this.rpcHook);//设置消费者组名this.rebalanceImpl.setConsumerGroup(this.defaultMQPushConsumer.getConsumerGroup());//设置消费模式this.rebalanceImpl.setMessageModel(this.defaultMQPushConsumer.getMessageModel());//设置分配mq策略（决定消费哪个mq的内容）       this.rebalanceImpl.setAllocateMessageQueueStrategy(this.defaultMQPushConsumer.getAllocateMessageQueueStrategy());//设置mq客户端实例对象this.rebalanceImpl.setmQClientFactory(this.mQClientFactory);//创建实际拉取消息的对象PullAPIWrapperthis.pullAPIWrapper = new PullAPIWrapper(mQClientFactory,this.defaultMQPushConsumer.getConsumerGroup(), isUnitMode());//注册消息过滤处理器列表this.pullAPIWrapper.registerFilterMessageHook(filterMessageHookList);//先在本地缓存中获取offsetstore，如果没有就去远程broker获取，offsetstore其实就是存放消费者消费mq的偏移量，防止重复消费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://集群模式// 集群模式下，用RemoteBrokerOffsetStore对象来处理消费偏移this.offsetStore = new RemoteBrokerOffsetStore(this.mQClientFactory, this.defaultMQPushConsumer.getConsumerGroup());break;default:break;}this.defaultMQPushConsumer.setOffsetStore(this.offsetStore);}// 加载本机偏移信息，集群模式用的是 RemoteBrokerOffsetStore 类对象，默认load没有实现内容，这是一个钩子函数，子类可自行实现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();
//将消费实例对象注册到本地内存consumerTable（ConcurrentMap）中，一个消费者组在一个客户端内，只会有一个消费实例对象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;}// 从namesrv中更新topic的路由信息this.updateTopicSubscribeInfoWhenSubscriptionChanged();// 通过Broker验证客户端的订阅数据合法性this.mQClientFactory.checkClientInBroker();// 发送心跳this.mQClientFactory.sendHeartbeatToAllBrokerWithLock();// 启动完成后，唤醒 RebalanceService 服务this.mQClientFactory.rebalanceImmediately();
}

checkConfig 检查配置项

private void checkConfig() throws MQClientException {// group名 合法性检查Validators.checkGroup(this.defaultMQPushConsumer.getConsumerGroup());// group判空，个人感觉这里的判断完全可以放在前面if (null == this.defaultMQPushConsumer.getConsumerGroup()) {throw new MQClientException("consumerGroup is null"+ FAQUrl.suggestTodo(FAQUrl.CLIENT_PARAMETER_CHECK_URL),null);}// group 不能是默认的名字if (this.defaultMQPushConsumer.getConsumerGroup().equals(MixAll.DEFAULT_CONSUMER_GROUP)) {throw new MQClientException("consumerGroup can not equal "+ MixAll.DEFAULT_CONSUMER_GROUP+ ", please specify another one."+ FAQUrl.suggestTodo(FAQUrl.CLIENT_PARAMETER_CHECK_URL),null);}// 消息模式不能为空，只有2种：集群和广播if (null == this.defaultMQPushConsumer.getMessageModel()) {throw new MQClientException("messageModel is null"+ FAQUrl.suggestTodo(FAQUrl.CLIENT_PARAMETER_CHECK_URL),null);}// 判断从哪开始消费，默认从最后一次偏移，包括最后一次偏移、最小偏移、最大偏移等if (null == this.defaultMQPushConsumer.getConsumeFromWhere()) {throw new MQClientException("consumeFromWhere is null"+ FAQUrl.suggestTodo(FAQUrl.CLIENT_PARAMETER_CHECK_URL),null);}// 时间格式化Date dt = UtilAll.parseDate(this.defaultMQPushConsumer.getConsumeTimestamp(), UtilAll.YYYYMMDDHHMMSS);if (null == dt) {throw new MQClientException("consumeTimestamp is invalid, the valid format is yyyyMMddHHmmss,but received "+ this.defaultMQPushConsumer.getConsumeTimestamp()+ " " + FAQUrl.suggestTodo(FAQUrl.CLIENT_PARAMETER_CHECK_URL), null);}// 消息队列分配策略，默认是平均分配if (null == this.defaultMQPushConsumer.getAllocateMessageQueueStrategy()) {throw new MQClientException("allocateMessageQueueStrategy is null"+ FAQUrl.suggestTodo(FAQUrl.CLIENT_PARAMETER_CHECK_URL),null);}// 订阅容器map，不为空if (null == this.defaultMQPushConsumer.getSubscription()) {throw new MQClientException("subscription is null"+ FAQUrl.suggestTodo(FAQUrl.CLIENT_PARAMETER_CHECK_URL),null);}// push方式是异步的，所以需要一个消息监听器，因此，这里不能为空if (null == this.defaultMQPushConsumer.getMessageListener()) {throw new MQClientException("messageListener is null"+ FAQUrl.suggestTodo(FAQUrl.CLIENT_PARAMETER_CHECK_URL),null);}// 顺序还是并发类型boolean orderly = this.defaultMQPushConsumer.getMessageListener() instanceof MessageListenerOrderly;boolean concurrently = this.defaultMQPushConsumer.getMessageListener() instanceof MessageListenerConcurrently;// 只能是二者选一，不能两者都不是if (!orderly && !concurrently) {throw new MQClientException("messageListener must be instanceof MessageListenerOrderly or MessageListenerConcurrently"+ FAQUrl.suggestTodo(FAQUrl.CLIENT_PARAMETER_CHECK_URL),null);}// 判断最小消费者线程数合法性，默认20if (this.defaultMQPushConsumer.getConsumeThreadMin() < 1|| this.defaultMQPushConsumer.getConsumeThreadMin() > 1000) {throw new MQClientException("consumeThreadMin Out of range [1, 1000]"+ FAQUrl.suggestTodo(FAQUrl.CLIENT_PARAMETER_CHECK_URL),null);}// 判断最大消费者线程数合法性，默认20if (this.defaultMQPushConsumer.getConsumeThreadMax() < 1 || this.defaultMQPushConsumer.getConsumeThreadMax() > 1000) {throw new MQClientException("consumeThreadMax Out of range [1, 1000]"+ FAQUrl.suggestTodo(FAQUrl.CLIENT_PARAMETER_CHECK_URL),null);}// consumeThreadMin 不可能比 consumeThreadMax 更大if (this.defaultMQPushConsumer.getConsumeThreadMin() > this.defaultMQPushConsumer.getConsumeThreadMax()) {throw new MQClientException("consumeThreadMin (" + this.defaultMQPushConsumer.getConsumeThreadMin() + ") "+ "is larger than consumeThreadMax (" + this.defaultMQPushConsumer.getConsumeThreadMax() + ")",null);}// 判断并发消费最大的偏移跨度，默认2000，非法证明消费过程有问题if (this.defaultMQPushConsumer.getConsumeConcurrentlyMaxSpan() < 1|| this.defaultMQPushConsumer.getConsumeConcurrentlyMaxSpan() > 65535) {throw new MQClientException("consumeConcurrentlyMaxSpan Out of range [1, 65535]"+ FAQUrl.suggestTodo(FAQUrl.CLIENT_PARAMETER_CHECK_URL),null);}// 队列级别的流量控制阈值，默认情况下每个消息队列最多缓存1000条消息，考虑{@code pullBatchSize}，瞬时值可能超过限制if (this.defaultMQPushConsumer.getPullThresholdForQueue() < 1 || this.defaultMQPushConsumer.getPullThresholdForQueue() > 65535) {throw new MQClientException("pullThresholdForQueue Out of range [1, 65535]"+ FAQUrl.suggestTodo(FAQUrl.CLIENT_PARAMETER_CHECK_URL),null);}// topic级别的流量控制阈值if (this.defaultMQPushConsumer.getPullThresholdForTopic() != -1) {if (this.defaultMQPushConsumer.getPullThresholdForTopic() < 1 || this.defaultMQPushConsumer.getPullThresholdForTopic() > 6553500) {throw new MQClientException("pullThresholdForTopic Out of range [1, 6553500]"+ FAQUrl.suggestTodo(FAQUrl.CLIENT_PARAMETER_CHECK_URL),null);}}// 队列级别的限制拉取大小，默认情况下每个消息队列最多缓存100Mif (this.defaultMQPushConsumer.getPullThresholdSizeForQueue() < 1 || this.defaultMQPushConsumer.getPullThresholdSizeForQueue() > 1024) {throw new MQClientException("pullThresholdSizeForQueue Out of range [1, 1024]"+ FAQUrl.suggestTodo(FAQUrl.CLIENT_PARAMETER_CHECK_URL),null);}// topic级别的限制拉取大小，默认情况下每个消息队列最多缓存100Mif (this.defaultMQPushConsumer.getPullThresholdSizeForTopic() != -1) {// pullThresholdSizeForTopicif (this.defaultMQPushConsumer.getPullThresholdSizeForTopic() < 1 || this.defaultMQPushConsumer.getPullThresholdSizeForTopic() > 102400) {throw new MQClientException("pullThresholdSizeForTopic Out of range [1, 102400]"+ FAQUrl.suggestTodo(FAQUrl.CLIENT_PARAMETER_CHECK_URL),null);}}// 拉取间隔时间if (this.defaultMQPushConsumer.getPullInterval() < 0 || this.defaultMQPushConsumer.getPullInterval() > 65535) {throw new MQClientException("pullInterval Out of range [0, 65535]"+ FAQUrl.suggestTodo(FAQUrl.CLIENT_PARAMETER_CHECK_URL),null);}// 批量消费消息的最大大小if (this.defaultMQPushConsumer.getConsumeMessageBatchMaxSize() < 1|| this.defaultMQPushConsumer.getConsumeMessageBatchMaxSize() > 1024) {throw new MQClientException("consumeMessageBatchMaxSize Out of range [1, 1024]"+ FAQUrl.suggestTodo(FAQUrl.CLIENT_PARAMETER_CHECK_URL),null);}// 批量拉取的大小if (this.defaultMQPushConsumer.getPullBatchSize() < 1 || this.defaultMQPushConsumer.getPullBatchSize() > 1024) {throw new MQClientException("pullBatchSize Out of range [1, 1024]"+ FAQUrl.suggestTodo(FAQUrl.CLIENT_PARAMETER_CHECK_URL),null);}
}

MQClientInstance.start

public void start() throws MQClientException {synchronized (this) {switch (this.serviceState) {case CREATE_JUST:this.serviceState = ServiceState.START_FAILED;//这一步就是获取name server的地址，正常都会指定的，否则没有什么意义if (null == this.clientConfig.getNamesrvAddr()) {this.mQClientAPIImpl.fetchNameServerAddr();}/* 启动 request-response channel，具体细节不在这里讲，以后有机会再写netty源码，主要做了以下工作1.设置netty工作线程2.设置netty收发消息时的处理器3.创建延迟3秒且每秒执行一次扫描过期无效的reponse信息的请求4.设置消息远程读取/接收的Netty处理器，NettyClientHandler，重点关注，先记着，后面会用*/this.mQClientAPIImpl.start();/* 启动rocketmq内置的一些定时任务，如下：1.延迟10秒且每2分钟执行一次获取最新name server 地址的任务2.延迟10毫秒且每{this.clientConfig.getPollNameServerInterval()	     （pollNameServerInterval可配置，但默认是30秒）}毫秒执行一次更新topic路由信息的任务MQClientInstance.this.updateTopicRouteInfoFromNameServer()3.延迟1秒且每{this.clientConfig.getHeartbeatBrokerInterval()	     （heartbeatBrokerInterval可配置，但默认是30秒）}毫秒执行一次清除下线的broker和发送心态到所有broker的任务4.延迟10秒且每{this.clientConfig.getPersistConsumerOffsetInterval()	     （persistConsumerOffsetInterval可配置，但默认是5秒）}毫秒执行一次持久化消息者offset值的任务，持久化存放在哪呢？放到broker中5.延迟1毫秒且每1分钟执行一次动态改变消息推送到消费者的工作线程数的任务*/this.startScheduledTask();//启动拉取消息服务this.pullMessageService.start();// 启动负载均衡服务this.rebalanceService.start();//因为MQClientInstance这个类，是生产者和消费都共用的，且消费者也会使用消费生产并发送操作，比如消费失败时重推消息，也需要启动生产者实现类，所以看到这段代码不用奇怪，如果是生产者，就算调用start方法，也不会启动2次，内部有针对状态的判断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;}}
}