从源码分析RocketMQ系列-Consumer消息接收逻辑

一时失言乱红尘 2023-07-16 06:57 11阅读 0赞

> **导语**  
>   在前面的分析中分析了关于Producer发送消息的逻辑，并且追踪到了在DefaultMQPushConsumerImpl 类中的有对应的消息监听方法，这个消息监听的方法是从Consumer调用 start()方法就开始启动的，但是从Consumer的示例代码中会看到，并没有像是Producer那样的明显表示Receive的方法而是通过我们下面要分析的方法来实现。

### 文章目录 ###

*   *  Consumer示例
     *   *  消息接收逻辑
         *  ConsumeMessageConcurrentlyService 类
         *  consumeMessageDirectly()
         *  方法调用链分析
         *  @ChannelHandler.Sharable说明
     *  总结

## Consumer示例 ##

下面就是在RocketMQ中提供的一段示例代码，可以看到大致分为三个步骤

*  1、Instantiate with specified consumer group name.
 *  2、Subscribe one more more topics to consume.
 *  3、Register callback to execute on arrival of messages fetched from brokers.
 *  3、Launch the consumer instance.

这里简单的来对比一下Producer的操作，如下

*  1、Instantiate with a producer group name.
 *  2、Launch the instance.
 *  3、Create a message instance, specifying topic, tag and message body.
 *  4、Call send message to deliver message to one of brokers.

通过上面的官方提示，对于Producer的SendMessage之前已经进行了追踪，那么Consumer端其他的操作在什么地方呢？

public class Consumer { 
    
        public static void main(String[] args) throws InterruptedException, MQClientException { 
    
            /* * Instantiate with specified consumer group name. */
            DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("please_rename_unique_group_name_4");
            consumer.setNamesrvAddr("localhost:9876");
            consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
    
            /* * Subscribe one more more topics to consume. */
    // consumer.subscribe("TopicTest", "*");
            consumer.subscribe("World", "*");
    
            consumer.registerMessageListener(new MessageListenerConcurrently() { 
    
                @Override
                public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
                    ConsumeConcurrentlyContext context) { 
                    System.out.printf("%s Receive New Messages: %s %n", Thread.currentThread().getName(), msgs);
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }
            });
    
            /* * Launch the consumer instance. */
            consumer.start();
    
            System.out.printf("Consumer Started.%n");
        }
    }

### 消息接收逻辑 ###

在上面代码中有一段逻辑，并且从日志的输出效果中我们会看到这段逻辑的不同之处，从下面截图中会看到，在Consumer测试类最后加了一段输出代码，按照正常的输出流程，这段代码会在所有的消息都发送完成之后才会输出。因为代码是从上到下执行的，但是有一种情况这段代码的输出不会影响其他逻辑，那就是多线程，其他线程不会影响主线程的正常执行。那么这段多线程在什么地方执行呢？  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25paHVpMTIz_size_16_color_FFFFFF_t_70]  
  通过输出结果的跟进，会看到实际上进行输出结果的是在下面这段代码中，这段代码consumer注册了一个MessageListener，消息监听者，并且传入了一个 并发消息监听。既然是监听，那它监听的肯定是一个服务，到底是个什么样的服务呢？继续来跟进  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25paHVpMTIz_size_16_color_FFFFFF_t_70 1]  
**跟进注册方法**  
  进入到注册方法中会看到，这个注册方法是由DefaultMQPushConsumer类来提供，并且传入的这个Listener被两个地方使用了

*  DefaultMQPushConsumer 类自身的 messageListener对象
 *  DefaultMQPushConsumerImpl 类的registerMessageListener()方法所使用  
    ![在这里插入图片描述][20200318102758941.png]  
      到这里，隐约的会觉得这个事情不是很简单。观察一下在注册的时候传入的对象其实是直接由接口实现的，而这个消息的读取也是直接从所定义的接口方法中来获取到的。在这个类的实现类中直接找到对应方法中并没有实际的调用逻辑  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25paHVpMTIz_size_16_color_FFFFFF_t_70 2]  
      通过查看类继承关系发现，直接使用匿名实现类的方式比较多一点，而实现类却是很少。  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25paHVpMTIz_size_16_color_FFFFFF_t_70 3]  
      通过上面的分析可以知道我们需要追踪的并不是MessageListenerConcurrently 接口的实现类。而是需要追踪它里面唯一的一个实现方法。如下。  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25paHVpMTIz_size_16_color_FFFFFF_t_70 4]  
      会发现这个方法被ConsumeMessageConcurrentlyService类中所调用。并且调用的对象是this.messageListener，从这里可以知道，其实MessageListener也实现了这个接口。真是这样么？其实并不是这样这个监听还是就是 MessageListenerConcurrently 对象;这里就来分析一下这个实现模式。

### ConsumeMessageConcurrentlyService 类 ###

Consumer消息并发服务类，这个类实现了一个接口ConsumeMessageService，这个接口有如下的几个方法

public interface ConsumeMessageService { 
        void start();
    
        void shutdown();
    
        void updateCorePoolSize(int corePoolSize);
    
        void incCorePoolSize();
    
        void decCorePoolSize();
    
        int getCorePoolSize();
    
        ConsumeMessageDirectlyResult consumeMessageDirectly(final MessageExt msg, final String brokerName);
    
        void submitConsumeRequest(
            final List<MessageExt> msgs,
            final ProcessQueue processQueue,
            final MessageQueue messageQueue,
            final boolean dispathToConsume);
    }

并且有如下的两个实现类，这两个实现类其中一个就是我们上面提到的。  
![在这里插入图片描述][2020031815291923.png]  
  其实可以看到，这个接口提供的规则里面最为重要就是最后的两个方法。

*  consumeMessageDirectly() 直接使用消息
 *  submitConsumeRequest() 提交Consumer请求

下面就来分析这两个方法在ConsumeMessageConcurrentlyService类中的实现

### consumeMessageDirectly() ###

进入这个方法之前先来分析这个方法的两个参数

*  MessageExt： 消息扩展类
 *  brokerName：Broker的Name，这个是通过配置文件中定义的

public ConsumeMessageDirectlyResult consumeMessageDirectly(MessageExt msg, String brokerName)

### 方法调用链分析 ###

在分析这个方法之前先来整理一下方法的调用链，通过方法调用链来找到具体整个的数据传递的流程。  
**接口方法第一次被调用**  
  根据方法的追踪接口方法this.messageListener.consumeMessage(),第一次被调用是在ConsumeMessageConcurrentlyService 类的consumeMessageDirectly()方法中。进入到该方法，发现如果要使用下面这个方法的话需要有人调用这个ConsumeMessageConcurrentlyService类中的方法，那么在什么地方进行调用的呢？继续跟进

ConsumeConcurrentlyStatus status = this.messageListener.consumeMessage(msgs, context);

**MQClientInstance中的consumeMessageDirectly()**  
  ConsumeMessageConcurrentlyService中的consumeMessageDirectly()的方法调用是被MQClientInstance类中的consumeMessageDirectly() 方法调用。是通过链式调用的方式，但是这里的链式调用并不需要太关心，应为既然调用了ConsumeMessageConcurrentlyService类中的方法那么它的上一步返回的就一定是对应的对象。这里我们关系在MQClientInstance中调用的方法。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25paHVpMTIz_size_16_color_FFFFFF_t_70 5]  
**ClientRemotingProcessor中的consumeMessageDirectly()**  
  在MQClientInstance中调用的方法是肯定是被其他跌俄方调用的，在ClientRemotingProcessor类中的consumeMessageDirectly()方法被调用了，跟之前一样，会看到，是一个MQClientInstance的对象，这个对象是Client和Server共同使用的。既然是ClientRemotingProcessor类中的方法就说明是Client的一个处理线程。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25paHVpMTIz_size_16_color_FFFFFF_t_70 6]  
**ClientRemotingProcessor的processRequest()方法**  
  继续跟进上面的方法，既然是一个线程处理方法，就需要有调用这个方法的入口，就追踪到了它的调用实现方法processRequest()。并且返回的是一个RemotingComand对象，这个对象之前说过的，被Request和Response共同使用。到这里会发现传入的参数都是一个Request为命名规则。

@Override
        public RemotingCommand processRequest(ChannelHandlerContext ctx,
            RemotingCommand request) throws RemotingCommandException { 
            switch (request.getCode()) { 
                case RequestCode.CHECK_TRANSACTION_STATE:
                    return this.checkTransactionState(ctx, request);
                case RequestCode.NOTIFY_CONSUMER_IDS_CHANGED:
                    return this.notifyConsumerIdsChanged(ctx, request);
                case RequestCode.RESET_CONSUMER_CLIENT_OFFSET:
                    return this.resetOffset(ctx, request);
                case RequestCode.GET_CONSUMER_STATUS_FROM_CLIENT:
                    return this.getConsumeStatus(ctx, request);
    
                case RequestCode.GET_CONSUMER_RUNNING_INFO:
                    return this.getConsumerRunningInfo(ctx, request);
    
                case RequestCode.CONSUME_MESSAGE_DIRECTLY:
                    return this.consumeMessageDirectly(ctx, request);
                default:
                    break;
            }
            return null;
        }

**NettyRemotingAbstract中的processRequestCommand方法调用**  
  追踪上面的request方法，进入到了一个Netty的抽象类，在这个抽象类中方法的调用是通过多线程的方式来进行的，这里看到了我们向往已久的东西，就是Response类，这个就是在Producer端我们分析的时候获取到的对象。也就是说从这里开始对象就进行了从Request到Response的交接。更为重要的一点是，这个多线程的调用。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25paHVpMTIz_size_16_color_FFFFFF_t_70 7]

上面方法的调用是被如下的一个方法调用，传入和ChannelHandlerContext以及Request参数，在一个什么样上下文中处理一个Request参数，这个方法的功能很明确。方法体中对消息的类型进行了判断。但是这个地方到底在什么地方被调用的呢。就到了最终的调用方

public void processMessageReceived(ChannelHandlerContext ctx, RemotingCommand msg) throws Exception

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25paHVpMTIz_size_16_color_FFFFFF_t_70 8]  
**最终调用**  
  会看到最终的调用被封装到了NettyClientHandler对象中，但是这个对象实现了SimpleChannelInboundHandler类中的方法，而NettyClientHandler类是一个内部类，分别在NettyRemotingServer和NettyRemotingClient都有定义，并且实现的方法都是一样的，但是在NettyRemotingServer端定义的时候有如下的一个注解@ChannelHandler.Sharable，对于这个注解在后面的分析中会做说明，它是由Netty提供的。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25paHVpMTIz_size_16_color_FFFFFF_t_70 9]

### @ChannelHandler.Sharable说明 ###

在上面的逻辑中出现了一个@ChannelHandler.Sharable注解，这个注解被Netty所提供。当某个ChannelInboundHandler的实现重写channelRead()方法时，它将负责显式地释放与池化的ByteBuf实例相关的内存。Netty为此提供了一个实用方法ReferenceCountUtil.release() 但是以这种方式管理资源可能很繁琐。  
  一个更加简单的方式是使用SimpleChannelInboundHandler。 由于SimpleChannelInboundHandler会自动释放资源，所以你不应该存储指向任何消息的引用供将来使用，因为这些引用都将会失效。而上面这个注解，标注一个channel handler可以被多个channel安全地共享。  
  ChannelHandlerAdapter还提供了实用方法isSharable()。如果其对应的实现被标注为Sharable，那么这个方法将返回true，表示它可以被添加到多个ChannelPipeline中。因为一个ChannelHandler可以从属于多个ChannelPipeline，所以它也可以绑定到多个ChannelHandlerContext实例。用于这种用法的ChannelHandler必须要使用@Sharable注解标注；否则，试图将它添加到多个ChannelPipeline时将会触发异常。显而易见，为了安全地被用于多个并发的Channel（即连接），这样的ChannelHandler必须是线程安全的。

到这里就可以理解为什么要使用这个注解了，在之前的Producer分析中再客户端start()方法中有这样一段代码，由于IDEA比较智能，它对未使用的对象所持有的态度就是不给它颜色。会看到这里的handler对象并没有被使用到，也就是说所有的对于这个对象的配置就只是简单的配置一下。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25paHVpMTIz_size_16_color_FFFFFF_t_70 10]  
  但是，在NettyRemotingServer类的实现中有下面这个方法，这个方法的操作是在start()方法中被调用。

private void prepareSharableHandlers() { 
            handshakeHandler = new HandshakeHandler(TlsSystemConfig.tlsMode);
            encoder = new NettyEncoder();
            connectionManageHandler = new NettyConnectManageHandler();
            serverHandler = new NettyServerHandler();
     }

&emps; 总结一下就是，所有的通过Channel的Connection创建以及Channel的创建都是来自于this.mQClientAPIImpl.start();方法而这个方法是被MQClientInstance中的start()方法调用，而MQClientInstance类则是把Producer 发送消息和 Consumer接收消息连接起来的关键类。

## 总结 ##

上面方法的整个的调用过程看上去很乱，但是实际上很简单，就只是将方法调用串联起来，并没有涉及到其中的一些其他参数以及配置的讨论，顺着上面的思路可以很好的了解Consumer的消息接收逻辑，总结一下就是从消息发送开始就已经准备开始接收了。其中有很多保证高效性的操作。在后续的分析中还会进行详解。这篇博客中最为关键的一点就是@ChannelHandler.Sharable注解的说明，以及this.mQClientAPIImpl.start()方法的使用时机。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25paHVpMTIz_size_16_color_FFFFFF_t_70]: /images/20230528/e6adaeb4d00340a4b0518b8b7a3350c5.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25paHVpMTIz_size_16_color_FFFFFF_t_70 1]: /images/20230528/00cc35c991f84d60bb9387b3e036ebf0.png
[20200318102758941.png]: /images/20230528/f575b16d0801433ab6b7e7e01d33bd56.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25paHVpMTIz_size_16_color_FFFFFF_t_70 2]: /images/20230528/78efe8b27d2e45ccbc8574e48df4374f.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25paHVpMTIz_size_16_color_FFFFFF_t_70 3]: /images/20230528/f2ca86169b064eae8b5b68c0d6e6f91b.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25paHVpMTIz_size_16_color_FFFFFF_t_70 4]: /images/20230528/632f5a3a272b4fcfb3a0b4709c0d43dc.png
[2020031815291923.png]: /images/20230528/3f56ced327a04d539bcc137c71140f95.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25paHVpMTIz_size_16_color_FFFFFF_t_70 5]: /images/20230528/f0f5ba60c58347138db1d04953f772ee.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25paHVpMTIz_size_16_color_FFFFFF_t_70 6]: /images/20230528/08a501a511754111977251af0eefdd93.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25paHVpMTIz_size_16_color_FFFFFF_t_70 7]: /images/20230528/d5d60d06ae804537a2076276f26c09e5.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25paHVpMTIz_size_16_color_FFFFFF_t_70 8]: /images/20230528/8bc3733853bc40c0836f34e4a47d358a.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25paHVpMTIz_size_16_color_FFFFFF_t_70 9]: /images/20230528/8773490e4cfb4f698b7925bf2573a948.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25paHVpMTIz_size_16_color_FFFFFF_t_70 10]: /images/20230528/dde2f082039b44418a86494c87a9aa2c.png