一、概述
一个最简单的Consumer的启动代码如下：
public static void main(String[] args) throws Exception{
// Instantiate with specified consumer group name.
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("c1");
// Specify name server addresses.
consumer.setNamesrvAddr("10.1.11.155:9876");
// Subscribe one more more topics to consume.
consumer.subscribe("qqq", "TagA||TagB");
// Register callback to execute on arrival of messages fetched from brokers.
consumer.registerMessageListener(new MessageListenerConcurrently() {
@Override
public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
ConsumeConcurrentlyContext context) {
msgs.forEach(m-> System.out.println(new String(m.getBody())));
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
}
});
//Launch the consumer instance.
consumer.start();
System.out.printf("Consumer Started.%n");
}
最重要的有几个步骤：
- new DefaultMQPushConsumer("c1");
- consumer.subscribe("qqq", "TagA||TagB");
- registerMessageListener()
-
start()
下面就这几个方法进行深入
二、实例化一个DefaultMQPushConsumer
通过下面的代码就能够实例化一个DefaultMQPushConsumer
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("c1");
这个默认构造方法其实调用了下面的构造方法
public DefaultMQPushConsumer(final String consumerGroup, RPCHook rpcHook,
AllocateMessageQueueStrategy allocateMessageQueueStrategy) {
this.consumerGroup = consumerGroup;
this.allocateMessageQueueStrategy = allocateMessageQueueStrategy;
defaultMQPushConsumerImpl = new DefaultMQPushConsumerImpl(this, rpcHook);
}
一共三个参数：
- consumerGroup
消费者组 - rpcHook
远程调用的hook，我看Rocket的德行一般就在RPC调用之前，针对访问控制（ACL）做了一个前置 - allocateMessageQueueStrategy
队列负载均衡策略，默认会用 AllocateMessageQueueAveragely（平均算法）
系统提供了6种策略：后续研究下- AllocateMachineRoomNearby
- AllocateMessageQueueAveragely
- AllocateMessageQueueAveragelyByCircle
- AllocateMessageQueueByConfig
- AllocateMessageQueueByMachineRoom
- AllocateMessageQueueConsistentHash
三、定义监听的主题及subExpression
代码如下：
consumer.subscribe("qqq", "TagA||TagB");
解析一下topic及subExpression，生成一个SubscriptionData实体（包含了topic、subExpression的value和hash值），并放到map中
四、注册监听处理器 registerMessageListener
consumer.registerMessageListener(new MessageListenerConcurrently() {
@Override
public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
ConsumeConcurrentlyContext context) {
// msgs.forEach(m-> System.out.println(new String(m.getBody())));
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
}
});
新建了MessageListenerConcurrently接口的匿名类，并实现了consumeMessage()方法，注意，这里MessageListenerConcurrently是一个接口，继承自MessageListener，同样继承自MessageListener的还有MessageListenerOrderly（顺序消费监听器）
五、启动 start()
这个是核心的核心了
1、this.checkConfig();
检查group等参数是否合法
2、this.copySubscription();
复制监听规则，其实就是把之前生成的SubscriptionData复制一份，交给RebalancePushImpl（这个应该是负载均衡消费的核心类），除了复制监听规则，还会把以下信息复制过去：
- setConsumerGroup 复制组名称
- setMessageModel 复制消费模式：广播/集群
- setAllocateMessageQueueStrategy 复制负载均衡算法
- setmQClientFactory（在下面3中初始化） 复制mqClient
3、实例化一个MQClientInstance
this.mQClientFactory = MQClientManager.getInstance().getAndCreateMQClientInstance(this.defaultMQPushConsumer, this.rpcHook);
4、指定offsetStore
如果是集群模式，则会实例化一个LocalFileOffsetStore
如果是广播模式，则会实例化一个RemoteBrokerOffsetStore
都会调用load的方法，但是RemoteBrokerOffsetStore是一个空实现
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();
5、实例化ConsumeMessageService
主要有两种：并发消费service及顺序消费service
- 并发消费 ConsumeMessageOrderlyService.start();
start() 会启动一个定时任务，延迟15分钟，每15分钟执行一次，清理失效的消息处理线程 - 顺序消费 ConsumeMessageConcurrentlyService
start() 会启动一个定时任务，延迟1秒，每20秒执行一次 lockMQPeriodically()方法，这个方法应该是顺序消费的精髓，可以深入研读一下
6、启动客户端 mQClientFactory.start();
producer启动的时候也会调用这个方法
上面1 ~ 5 的步骤，大部分是为了这个start()做准备，这个start做了很多事情，如下：
-
this.mQClientAPIImpl.fetchNameServerAddr();
如果没有指定namesrv的地址，则会通过RPC获取，这个最终是通过调用一个外部的URL连接来获取的，这个链接可以自己定义，官方也推荐这种方法，最底层也最灵活。 -
this.mQClientAPIImpl.start();
启动netty客户端，可以和broker及namesrv通讯的通道 -
this.startScheduledTask();
启动一批自动任务-
MQClientInstance.this.mQClientAPIImpl.fetchNameServerAddr();
定期获取namesrv的地址，前面在启动的时候已经获取一次了 -
MQClientInstance.this.updateTopicRouteInfoFromNameServer();
定期从namesrv拉取meta信息及topic等路由信息 -
cleanOfflineBroker()、sendHeartbeatToAllBrokerWithLock()
定期检查下线的broker及发送心跳 -
MQClientInstance.this.persistAllConsumerOffset();
定期固化消费者的offset -
MQClientInstance.this.adjustThreadPool();
定期调整处理线程池的大小，这个4.4版本好像是个空实现，没用
-
MQClientInstance.this.mQClientAPIImpl.fetchNameServerAddr();
-
this.pullMessageService.start();
启动客户端主动pull的服务 -
this.rebalanceService.start();
启动负载均衡服务
最后还会调用一次getDefaultMQProducerImpl().start(false)，没错，就是producer的start，但是参数是false，不会最终调用producer的start，就把配置刷过去了。
7、其他方法
- this.updateTopicSubscribeInfoWhenSubscriptionChanged();
- this.mQClientFactory.checkClientInBroker();
- this.mQClientFactory.sendHeartbeatToAllBrokerWithLock();
-
this.mQClientFactory.rebalanceImmediately();
这几个方法后续研究下
到此为止，consumer的启动流程完毕
