RocketMQ之Consumer整体介绍启动源码分析

// org.apache.rocketmq.client.consumer.MQConsumer

public interface MQConsumer extends MQAdmin {

     // 如果消费失败，消息将被发送回代理，并延迟消耗一些时间

     void sendMessageBack( final MessageExt msg /*消息*/ , final int delayLevel /*延迟级别*/ , final String brokerName); 

     // 根据topic从使用者缓存中获取消息队列

     Set<MessageQueue> fetchSubscribeMessageQueues( final String topic) throws MQClientException;

}

// org.apache.rocketmq.client.consumer.MQPushConsumer

public interface MQPushConsumer extends MQConsumer {

     // 启动Consumer

     void start() throws MQClientException;

     // 关闭Consumer

     void shutdown();

     // 注册并发消息Listener

     void registerMessageListener( final MessageListenerConcurrently messageListener);

     // 注册顺序消息Listener,将会有序地接收消息。一个队列一个线程

     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的默认实现，DefaultMQPushConsumer中大部分功能都是对它的代理

     protected final transient DefaultMQPushConsumerImpl defaultMQPushConsumerImpl;

     // 相同角色的消费者需要具有完全相同的subscriptions和consumerGroup才能正确实现负载平衡，它需要全局唯一

     private String consumerGroup;

     // 消息模型定义了如何将消息传递到每个消费者客户端的方式，默认是集群模式

     private MessageModel messageModel = MessageModel.CLUSTERING;

     // 第一次消费时指定的消费策略，默认是CONSUME_FROM_LAST_OFFSET

     private ConsumeFromWhere consumeFromWhere = ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET;

     // 队列分配算法，指定如何将消息队列分配给每个使用者客户端。

     private AllocateMessageQueueStrategy allocateMessageQueueStrategy;

     // 订阅关系

     private Map<String /* topic */ , String /* sub expression */ > subscription = new HashMap<String, String>();

     // 消息监听器

     private MessageListener messageListener;

     // 消息消费进度存储器

     private OffsetStore offsetStore;

     // 最小消费线程数

     private int consumeThreadMin = 20 ;

     // 最大消费线程数

     private int consumeThreadMax = 20 ;

     // 推送模式下拉去消息的间隔时间，默认一次拉取消息完成后立刻继续拉取

     private long pullInterval = 0 ;

     // 批量消费数量

     private int consumeMessageBatchMaxSize = 1 ;

     // 批量拉取的数量

     private int pullBatchSize = 32 ;

     // 每次拉取时是否更新订阅关系,默认是false

     private boolean postSubscriptionWhenPull = false ;

     // 消息最大重试次数，如果消息消费最大次数超过maxReconsumeTimes还未成功，则消息将被转移到一个失败队列

     private int maxReconsumeTimes = - 1 ;

     //延迟将该队列的消息提交到消费者线程的等待时间，默认延迟1s

     private long suspendCurrentQueueTimeMillis = 1000 ;

     // 消息阻塞消费线程的最大超时时间，默认15分钟

     private long consumeTimeout = 15 ;

     // 关闭使用者时等待消息的最长时间，0表示没有等待。

     private long awaitTerminationMillisWhenShutdown = 0 ;

}

public static void main(String[] args) throws MQClientException {

     // 1. 传入CONSUMER_GROUP，创建DefaultMQPushConsumer

     DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(CONSUMER_GROUP);

     // 2. 设置namesrvAddr

     consumer.setNamesrvAddr( "127.0.0.1:9876" );

     // 3. 订阅Topic

     consumer.subscribe(TOPIC, "*" );

     // 4.注册消息Listener

     consumer.registerMessageListener((MessageListenerConcurrently) (msg, context) -&gt; {

         System.out.printf( "%s Receive New Messages: %s %n" , Thread.currentThread().getName(), msg);

         return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;

     });

     // 5.启动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启动

     this .defaultMQPushConsumerImpl.start();

     // 消息轨迹跟踪服务，默认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 {

   //...

   // 状态先设置为启动失败

   this .serviceState = ServiceState.START_FAILED;

   // 校验配置，ConsumerGroup校验，

   this .checkConfig();

   // 订阅关系copy到RebalanceImpl中

   this .copySubscription();

   // 如果是集群模式，消费者名称如果是DEFAULT，则会改成：PID#时间戳

   if ( this .defaultMQPushConsumer.getMessageModel() == MessageModel.CLUSTERING) {

       this .defaultMQPushConsumer.changeInstanceNameToPID();

   }

   //...

}

// org.apache.rocketmq.client.impl.consumer.DefaultMQPushConsumerImpl#start

// 生成一个MQClientInstance

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());

// 设置MQClientInstance

this .rebalanceImpl.setmQClientFactory( this .mQClientFactory);

// 构建拉消息包装器

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存储偏移量

             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);

}

// 如果是广播模式，则从本地文件load偏移量，如果是集群模式则是一个空实现

this .offsetStore.load();

// org.apache.rocketmq.client.impl.consumer.DefaultMQPushConsumerImpl#start

// 如果是顺序消费

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

boolean registerOK = mQClientFactory.registerConsumer( this .defaultMQPushConsumer.getConsumerGroup(), this );

// ...

mQClientFactory.start();

// org.apache.rocketmq.client.impl.consumer.DefaultMQPushConsumerImpl#start

// 向Namesrv拉取并更新当前消费者订阅topic路由信息

this .updateTopicSubscribeInfoWhenSubscriptionChanged();

// 随机选择一个Broker，发送检查客户端tag配置的请求，主要是检测Broker是否支持SQL92类型的tag过滤以及SQL92的tag语法是否正确

this .mQClientFactory.checkClientInBroker();

// 给所有Broker发送心跳

this .mQClientFactory.sendHeartbeatToAllBrokerWithLock();

// 唤醒负载均衡服务rebalanceService，并进行rebalance

this .mQClientFactory.rebalanceImmediately();