RocketMQ之Pull消费者客户端启动

Pull消费者客户端(主动拉取消息的消费者)即构造了DefaultMQPullConsumer对象，DefaultMQPullConsumer继承了ClientConfig类。我们先看其构造方法

1[java] view plain copy
2public DefaultMQPullConsumer(final String consumerGroup, RPCHook rpcHook) {  
3    this.consumerGroup = consumerGroup;  
4    defaultMQPullConsumerImpl = new DefaultMQPullConsumerImpl(this, rpcHook);  
5}

这里只是简单设置了consumerGroup消费者组名，表示消费者属于哪个组。构造了DefaultMQPullConsumerImpl的实例，DefaultMQPullConsumerImpl的构造方法很简单，只是绑定了DefaultMQPullConsumer、配置了传入的rpcHook。

DefaultMQPullConsumer内部封装了DefaultMQPullConsumerImpl，其中还维护这一些配置信息。这里维护着消费者订阅的topic集合。

1[java] view plain copy
2private Set<String> registerTopics = new HashSet<String>();

整个消费者客户端的启动，调用了DefaultMQPullConsumer的start()方法，内部直接调用DefaultMQPullConsumerImpl的start()方法，这个start方法加了synchronized修饰。

1[java] view plain copy
2    public synchronized void start() throws MQClientException {  
3        switch (this.serviceState) {  
4            case CREATE_JUST:  
5                this.serviceState = ServiceState.START_FAILED;  
6  
7                this.checkConfig();  
8  
9                this.copySubscription();  
10  
11                if (this.defaultMQPullConsumer.getMessageModel() == MessageModel.CLUSTERING) {  
12                    this.defaultMQPullConsumer.changeInstanceNameToPID();  
13                }  
14  
15                this.mQClientFactory = MQClientManager.getInstance().getAndCreateMQClientInstance(this.defaultMQPullConsumer  
16                                                                                                                , this.rpcHook);  
17  
18                this.rebalanceImpl.setConsumerGroup(this.defaultMQPullConsumer.getConsumerGroup());  
19                this.rebalanceImpl.setMessageModel(this.defaultMQPullConsumer.getMessageModel());  
20                this.rebalanceImpl.setAllocateMessageQueueStrategy(this.defaultMQPullConsumer.getAllocateMessageQueueStrategy());  
21                this.rebalanceImpl.setmQClientFactory(this.mQClientFactory);  
22  
23                this.pullAPIWrapper = new PullAPIWrapper(  
24                    mQClientFactory,  
25                    this.defaultMQPullConsumer.getConsumerGroup(), isUnitMode());  
26                this.pullAPIWrapper.registerFilterMessageHook(filterMessageHookList);  
27  
28                if (this.defaultMQPullConsumer.getOffsetStore() != null) {  
29                    this.offsetStore = this.defaultMQPullConsumer.getOffsetStore();  
30                } else {  
31                    switch (this.defaultMQPullConsumer.getMessageModel()) {  
32                        case BROADCASTING:  
33                            this.offsetStore = new LocalFileOffsetStore(this.mQClientFactory, this.defaultMQPullConsumer  
34                                                                                                        .getConsumerGroup());  
35                            break;  
36                        case CLUSTERING:  
37                            this.offsetStore = new RemoteBrokerOffsetStore(this.mQClientFactory, this.defaultMQPullConsumer  
38                                                                                                        .getConsumerGroup());  
39                            break;  
40                        default:  
41                            break;  
42                    }  
43                    this.defaultMQPullConsumer.setOffsetStore(this.offsetStore);  
44                }  
45  
46                this.offsetStore.load();  
47  
48                boolean registerOK = mQClientFactory.registerConsumer(this.defaultMQPullConsumer.getConsumerGroup(), this);  
49                if (!registerOK) {  
50                    this.serviceState = ServiceState.CREATE_JUST;  
51  
52                    throw new MQClientException("The consumer group[" + this.defaultMQPullConsumer.getConsumerGroup()  
53                        + "] has been created before, specify another name please." + FAQUrl.suggestTodo(FAQUrl  
54                                                                                            .GROUP_NAME_DUPLICATE_URL), null);  
55                }  
56  
57                mQClientFactory.start();  
58                log.info("the consumer [{}] start OK", this.defaultMQPullConsumer.getConsumerGroup());  
59                this.serviceState = ServiceState.RUNNING;  
60                break;  
61            case RUNNING:  
62            case START_FAILED:  
63            case SHUTDOWN_ALREADY:  
64                throw new MQClientException("The PullConsumer service state not OK, maybe started once, "  
65                    + this.serviceState  
66                    + FAQUrl.suggestTodo(FAQUrl.CLIENT_SERVICE_NOT_OK),  
67                    null);  
68            default:  
69                break;  
70        }  
71  
72    }

一开始的serverState的状态自然为CREAT_JUST，调用checkConfig()，其中先是对ConsumerGroup进行验证，非空，合法(符合正则规则，且长度不超过配置最大值)，且不为默认值(防止消费者集群名冲突)，然后对消费者消息模式、消息队列分配算法进行非空、合法校验。

关于消费者消息模式有BroadCasting(广播)跟Clustering(集群)两种、默认是Clustering(集群)配置在DefaultMQPullConsumer中。关于消费者的消息分配算法，在DefaultMQPullConsumer中实现有默认的消息分配算法，allocateMessageQueueStrategy = new AllocateMessageQueueAveragely();(平均分配算法)。其实现了AllocateMessageQueueStrategy接口，重点看其实现的allocate()方法。

1[java] view plain copy
2@Override  
3public List<MessageQueue> allocate(String consumerGroup, String currentCID, List<MessageQueue> mqAll,  
4    List<String> cidAll) {  
5    if (currentCID == null || currentCID.length() < 1) {  
6        throw new IllegalArgumentException("currentCID is empty");  
7    }  
8    if (mqAll == null || mqAll.isEmpty()) {  
9        throw new IllegalArgumentException("mqAll is null or mqAll empty");  
10    }  
11    if (cidAll == null || cidAll.isEmpty()) {  
12        throw new IllegalArgumentException("cidAll is null or cidAll empty");  
13    }  
14  
15    List<MessageQueue> result = new ArrayList<MessageQueue>();  
16    if (!cidAll.contains(currentCID)) {  
17        log.info("[BUG] ConsumerGroup: {} The consumerId: {} not in cidAll: {}",  
18            consumerGroup,  
19            currentCID,  
20            cidAll);  
21        return result;  
22    }  
23  
24    int index = cidAll.indexOf(currentCID);  
25    int mod = mqAll.size() % cidAll.size();  
26    int averageSize =  
27        mqAll.size() <= cidAll.size() ? 1 : (mod > 0 && index < mod ? mqAll.size() / cidAll.size()  
28            + 1 : mqAll.size() / cidAll.size());  
29    int startIndex = (mod > 0 && index < mod) ? index * averageSize : index * averageSize + mod;  
30    int range = Math.min(averageSize, mqAll.size() - startIndex);  
31    for (int i = 0; i < range; i++) {  
32        result.add(mqAll.get((startIndex + i) % mqAll.size()));  
33    }  
34    return result;  
35}

传入的参数有当前消费者id，所有消息队列数组，以及当前所有消费者数组。先简单验证非空，再通过消费者数组大小跟消息队列大小根据平均算法算出当前消费者该分配哪些消息队列集合。逻辑不难。RocketMQ还提供了循环平均、一致性哈希、配置分配等算法，这里默认采用平均分配。

我们再回到DefaultMQPullConsumerImpl的start()方法，checkConfig后，调用copySubscription()方法，将配置在DefaultMQPullConsumer中的topic信息构造成并构造成subscriptionData数据结构，以topic为key以subscriptionData为value以键值对形式存到rebalanceImpl的subscriptionInner中。

1[java] view plain copy
2private void copySubscription() throws MQClientException {  
3    try {  
4        Set<String> registerTopics = this.defaultMQPullConsumer.getRegisterTopics();  
5        if (registerTopics != null) {  
6            for (final String topic : registerTopics) {  
7                SubscriptionData subscriptionData = FilterAPI.buildSubscriptionData(this.defaultMQPullConsumer.getConsumerGroup(),  
8                    topic, SubscriptionData.SUB_ALL);  
9                this.rebalanceImpl.getSubscriptionInner().put(topic, subscriptionData);  
10            }  
11        }  
12    } catch (Exception e) {  
13        throw new MQClientException("subscription exception", e);  
14    }  
15}

接下来从MQCLientManager中得到MQClient的实例，这个步骤跟生产者客户端相同。

再往后是对rebalanceImpl的配置，我们重点看下rebalanceImpl，它是在DefaultMQPullConsumerImpl成员中直接构造private RebalanceImpl rebalanceImpl = new RebalancePullImpl(this);即在DefaultMQPullConsumerImpl初始化的时候构造。接下来对其消费者组名、消息模式(默认集群)、队列分配算法(默认平均分配)、消费者客户端实例进行配置，配置信息都是从DefaultMQPullConsumer中取得。

1[java] view plain copy
2public abstract class RebalanceImpl {  
3    protected static final Logger log = ClientLogger.getLog();  
4    protected final ConcurrentMap<MessageQueue, ProcessQueue> processQueueTable = new ConcurrentHashMap<MessageQueue, ProcessQueue>(64);  
5    protected final ConcurrentMap<String/* topic */, Set<MessageQueue>> topicSubscribeInfoTable =  
6        new ConcurrentHashMap<String, Set<MessageQueue>>();  
7    protected final ConcurrentMap<String /* topic */, SubscriptionData> subscriptionInner =  
8        new ConcurrentHashMap<String, SubscriptionData>();  
9    protected String consumerGroup;  
10    protected MessageModel messageModel;  
11    protected AllocateMessageQueueStrategy allocateMessageQueueStrategy;  
12    protected MQClientInstance mQClientFactory;

接下来构造了PullAPIWrapper，仅仅调用其构造方法，简单的配置下

1[java] view plain copy
2public PullAPIWrapper(MQClientInstance mQClientFactory, String consumerGroup, boolean unitMode) {  
3    this.mQClientFactory = mQClientFactory;  
4    this.consumerGroup = consumerGroup;  
5    this.unitMode = unitMode;  
6}

然后初始化消费者的offsetStore，offset即偏移量，可以理解为消费进度，这里根据不同的消息模式来选择不同的策略。如果是广播模式，那么所有消费者都应该收到订阅的消息，那么每个消费者只应该自己消费的消费队列的进度，那么需要把消费进度即offsetStore存于本地采用LocalFileOffsetStroe，相反的如果是集群模式，那么集群中的消费者来平均消费消息队列，那么应该把消费进度存于远程采用RemoteBrokerOffsetStore。然后调用相应的load方法加载。

之后将当前消费者注册在MQ客户端实例上之后，调用MQClientInstance的start()方法，启动消费者客户端。

1[java] view plain copy
2    public void start() throws MQClientException {  
3  
4        synchronized (this) {  
5            switch (this.serviceState) {  
6                case CREATE_JUST:  
7                    this.serviceState = ServiceState.START_FAILED;  
8                    // If not specified,looking address from name server  
9                    if (null == this.clientConfig.getNamesrvAddr()) {  
10                        this.mQClientAPIImpl.fetchNameServerAddr();  
11                    }  
12                    // Start request-response channel  
13                    this.mQClientAPIImpl.start();  
14                    // Start various schedule tasks  
15                    this.startScheduledTask();  
16                    // Start pull service  
17                    this.pullMessageService.start();  
18                    // Start rebalance service  
19                    this.rebalanceService.start();  
20                    // Start push service  
21                    this.defaultMQProducer.getDefaultMQProducerImpl().start(false);  
22                    log.info("the client factory [{}] start OK", this.clientId);  
23                    this.serviceState = ServiceState.RUNNING;  
24                    break;  
25                case RUNNING:  
26                    break;  
27                case SHUTDOWN_ALREADY:  
28                    break;  
29                case START_FAILED:  
30                    throw new MQClientException("The Factory object[" + this.getClientId() + "] has been created before, and failed."  
31                                                                                                                                , null);  
32                default:  
33                    break;  
34            }  
35        }  
36    }

看到这里应该很熟悉，跟生产者客户端这里是同一段代码，无非解析路由消息并完成路由消息的配置，启动netty客户端，启动定时任务(定时更新从名称服务器获取路由信息更新本地路由信息，心跳，调整线程数量)，后面启动pull server、rebalance service、push service最后把serviceState状态设为Running表示客户端启动。

我们在这里重点看下RebalanceService的启动。下面贴出的是RebalanceService的run()方法。

1[java] view plain copy
2@Override  
3public void run() {  
4    log.info(this.getServiceName() + " service started");  
5  
6    while (!this.isStopped()) {  
7        this.waitForRunning(waitInterval);  
8        this.mqClientFactory.doRebalance();  
9    }  
10  
11    log.info(this.getServiceName() + " service end");  
12}

可以看到，只要这个线程没有被停止(客户端没关闭)，会一直循环调用客户端的doRebalance()方法。

1[java] view plain copy
2public void doRebalance() {  
3    for (Map.Entry<String, MQConsumerInner> entry : this.consumerTable.entrySet()) {  
4        MQConsumerInner impl = entry.getValue();  
5        if (impl != null) {  
6            try {  
7                impl.doRebalance();  
8            } catch (Throwable e) {  
9                log.error("doRebalance exception", e);  
10            }  
11        }  
12    }  
13}  

MQClientInstance遍历consumerTable(之前注册的时候以consumerGroup为key，以消费者客户端DefaultMQPullConsumerImpl为value存入consumerTable中)中的每个元素，循环调用其元素的doRebalance()方法。那我们看DefaultMQPullConsumerImpl的doRebalance方法。

11 [java] view plain copy
22 @Override  
33 public void doRebalance() {  
44     if (this.rebalanceImpl != null) {  
55         this.rebalanceImpl.doRebalance(false);  
66     }  
77 }

直接调用了rebalanceImpl的doRebalance方法

1[java] view plain copy
2public void doRebalance(final boolean isOrder) {  
3    Map<String, SubscriptionData> subTable = this.getSubscriptionInner();  
4    if (subTable != null) {  
5        for (final Map.Entry<String, SubscriptionData> entry : subTable.entrySet()) {  
6            final String topic = entry.getKey();  
7            try {  
8                this.rebalanceByTopic(topic, isOrder);  
9            } catch (Throwable e) {  
10                if (!topic.startsWith(MixAll.RETRY_GROUP_TOPIC_PREFIX)) {  
11                    log.warn("rebalanceByTopic Exception", e);  
12                }  
13            }  
14        }  
15    }  
16  
17    this.truncateMessageQueueNotMyTopic();  
18}  

可以看到先得到subTable即subscriptionInner，之前根据配置的每个topic生成的SubscriptionData数据结构的map。先遍历该map，得到每个topic，针对每个topic调用rebalanceByTopic()

1 1 [java] view plain copy
2 2     private void rebalanceByTopic(final String topic, final boolean isOrder) {  
3 3         switch (messageModel) {  
4 4             case BROADCASTING: {  
5 5                 Set<MessageQueue> mqSet = this.topicSubscribeInfoTable.get(topic);  
6 6                 if (mqSet != null) {  
7 7                     boolean changed = this.updateProcessQueueTableInRebalance(topic, mqSet, isOrder);  
8 8                     if (changed) {  
9 9                         this.messageQueueChanged(topic, mqSet, mqSet);  
1010                         log.info("messageQueueChanged {} {} {} {}",  
1111                             consumerGroup,  
1212                             topic,  
1313                             mqSet,  
1414                             mqSet);  
1515                     }  
1616                 } else {  
1717                     log.warn("doRebalance, {}, but the topic[{}] not exist.", consumerGroup, topic);  
1818                 }  
1919                 break;  
2020             }  
2121             case CLUSTERING: {  
2222                 Set<MessageQueue> mqSet = this.topicSubscribeInfoTable.get(topic);  
2323                 List<String> cidAll = this.mQClientFactory.findConsumerIdList(topic, consumerGroup);  
2424                 if (null == mqSet) {  
2525                     if (!topic.startsWith(MixAll.RETRY_GROUP_TOPIC_PREFIX)) {  
2626                         log.warn("doRebalance, {}, but the topic[{}] not exist.", consumerGroup, topic);  
2727                     }  
2828                 }  
2929   
3030                 if (null == cidAll) {  
3131                     log.warn("doRebalance, {} {}, get consumer id list failed", consumerGroup, topic);  
3232                 }  
3333   
3434                 if (mqSet != null && cidAll != null) {  
3535                     List<MessageQueue> mqAll = new ArrayList<MessageQueue>();  
3636                     mqAll.addAll(mqSet);  
3737   
3838                     Collections.sort(mqAll);  
3939                     Collections.sort(cidAll);  
4040   
4141                     AllocateMessageQueueStrategy strategy = this.allocateMessageQueueStrategy;  
4242   
4343                     List<MessageQueue> allocateResult = null;  
4444                     try {  
4545                         allocateResult = strategy.allocate(  
4646                             this.consumerGroup,  
4747                             this.mQClientFactory.getClientId(),  
4848                             mqAll,  
4949                             cidAll);  
5050                     } catch (Throwable e) {  
5151                         log.error("AllocateMessageQueueStrategy.allocate Exception. allocateMessageQueueStrategyName={}",  
5252                              strategy.getName(), e);  
5353                         return;  
5454                     }  
5555   
5656                     Set<MessageQueue> allocateResultSet = new HashSet<MessageQueue>();  
5757                     if (allocateResult != null) {  
5858                         allocateResultSet.addAll(allocateResult);  
5959                     }  
6060   
6161                     boolean changed = this.updateProcessQueueTableInRebalance(topic, allocateResultSet, isOrder);  
6262                     if (changed) {  
6363                         log.info(  
6464                             "rebalanced result changed. allocateMessageQueueStrategyName={}, group={}, topic={}, clientId={}, mqAllSize={}  
6565                         , cidAllSize={}, rebalanceResultSize={}, rebalanceResultSet={}",  
6666                             strategy.getName(), consumerGroup, topic, this.mQClientFactory.getClientId(), mqSet.size(), cidAll.size(),  
6767                             allocateResultSet.size(), allocateResultSet);  
6868                         this.messageQueueChanged(topic, mqSet, allocateResultSet);  
6969                     }  
7070                 }  
7171                 break;  
7272             }  
7373             default:  
7474                 break;  
7575         }  
7676     }

我们先重点关注集群模式下，先得到topic的本地路由信息，再通过topic跟这个消费者的组名，调用netty客户端的同步网络访问topic指定的broker，从broker端得到与其连接的且是指定消费组名下订阅指定topic的消费者id的集合。然后采用默认的分配算法的allocate()进行队列给消费者平均分配。然后调用updateProcessQueueTableInRebalance()方法判断是否重新队列分配。

1 1 [java] view plain copy
2 2 private boolean updateProcessQueueTableInRebalance(final String topic, final Set<MessageQueue> mqSet,  
3 3     final boolean isOrder) {  
4 4     boolean changed = false;  
5 5   
6 6     Iterator<Entry<MessageQueue, ProcessQueue>> it = this.processQueueTable.entrySet().iterator();  
7 7     while (it.hasNext()) {  
8 8         Entry<MessageQueue, ProcessQueue> next = it.next();  
9 9         MessageQueue mq = next.getKey();  
1010         ProcessQueue pq = next.getValue();  
1111   
1212         if (mq.getTopic().equals(topic)) {  
1313             if (!mqSet.contains(mq)) {  
1414                 pq.setDropped(true);  
1515                 if (this.removeUnnecessaryMessageQueue(mq, pq)) {  
1616                     it.remove();  
1717                     changed = true;  
1818                     log.info("doRebalance, {}, remove unnecessary mq, {}", consumerGroup, mq);  
1919                 }  
2020             } else if (pq.isPullExpired()) {  
2121                 switch (this.consumeType()) {  
2222                     case CONSUME_ACTIVELY:  
2323                         break;  
2424                     case CONSUME_PASSIVELY:  
2525                         pq.setDropped(true);  
2626                         if (this.removeUnnecessaryMessageQueue(mq, pq)) {  
2727                             it.remove();  
2828                             changed = true;  
2929                             log.error("[BUG]doRebalance, {}, remove unnecessary mq, {}, because pull is pause, so try to fixed it",  
3030                                 consumerGroup, mq);  
3131                         }  
3232                         break;  
3333                     default:  
3434                         break;  
3535                 }  
3636             }  
3737         }  
3838     }  
3939   
4040     List<PullRequest> pullRequestList = new ArrayList<PullRequest>();  
4141     for (MessageQueue mq : mqSet) {  
4242         if (!this.processQueueTable.containsKey(mq)) {  
4343             if (isOrder && !this.lock(mq)) {  
4444                 log.warn("doRebalance, {}, add a new mq failed, {}, because lock failed", consumerGroup, mq);  
4545                 continue;  
4646             }  
4747   
4848             this.removeDirtyOffset(mq);  
4949             ProcessQueue pq = new ProcessQueue();  
5050             long nextOffset = this.computePullFromWhere(mq);  
5151             if (nextOffset >= 0) {  
5252                 ProcessQueue pre = this.processQueueTable.putIfAbsent(mq, pq);  
5353                 if (pre != null) {  
5454                     log.info("doRebalance, {}, mq already exists, {}", consumerGroup, mq);  
5555                 } else {  
5656                     log.info("doRebalance, {}, add a new mq, {}", consumerGroup, mq);  
5757                     PullRequest pullRequest = new PullRequest();  
5858                     pullRequest.setConsumerGroup(consumerGroup);  
5959                     pullRequest.setNextOffset(nextOffset);  
6060                     pullRequest.setMessageQueue(mq);  
6161                     pullRequest.setProcessQueue(pq);  
6262                     pullRequestList.add(pullRequest);  
6363                     changed = true;  
6464                 }  
6565             } else {  
6666                 log.warn("doRebalance, {}, add new mq failed, {}", consumerGroup, mq);  
6767             }  
6868         }  
6969     }  
7070   
7171     this.dispatchPullRequest(pullRequestList);  
7272   
7373     return changed;  
7474 }

先遍历processQueueTable，看其topic下的该处理消息队列是否还是应该处理，由于新分配之后，消息队列可能会改变，所以原该处理的消息队列可能没必要处理，因此没必要处理的消息队列移除。当然也有可能多出需要处理的消息队列，于是需要建立其与processQueue的对应关系，先调用computerPullFromWhere得到该条消息下次拉取数据的位置，在RebalancePullImpl中实现了该方法直接返回0，把该处理的mq封装成pq后，更新到processQueueTable中。若有更新，无论是增加还是删除，则changed都设为true。(这个地方讲的有点模糊，他是客户端pull与push区别的关键，实际上push不过是在pull之上封装了下操作，后面我们会重新回来分析。)

方法返回后，如果changed为true，会调用messageQueueChanged方法来通知配置在DefaultMQPullConsumer中的相关messageQueueListener，我们可以看到RebalancePullImpl中的实现。

1 1 [java] view plain copy
2 2 @Override  
3 3 public void messageQueueChanged(String topic, Set<MessageQueue> mqAll, Set<MessageQueue> mqDivided) {  
4 4     MessageQueueListener messageQueueListener = this.defaultMQPullConsumerImpl.getDefaultMQPullConsumer().getMessageQueueListener();  
5 5     if (messageQueueListener != null) {  
6 6         try {  
7 7             messageQueueListener.messageQueueChanged(topic, mqAll, mqDivided);  
8 8         } catch (Throwable e) {  
9 9             log.error("messageQueueChanged exception", e);  
1010         }  
1111     }  
1212 }

广播模式则比较简单，由于所有消费者都要处理，少了队列分配这个步骤。

本文转载自:https://blog.csdn.net/panxj856856/article/details/80725630