从源码分析RocketMQ系列-MQClientInstance类详解

旧城等待， 2023-07-17 06:30 95阅读 0赞

> **导语**  
>   在之前的分析中，看到有一个类MQClientInstance，这个无论是在Producer端还是在Consumer端都是很重要的一个类，很多的功能都是从这个类发起的，这边分享中就来详细的看看这个类的功能。

### 文章目录 ###

*   *   *  基础对象
         *  构造方法
         *  配置信息
         *  加锁操作
         *  服务操作
         *  实际处理操作
     *  总结

首先要知道这个类的功应该是整个的RocketMQ的消息发送的基础，有了这个基础之后后续的内容才能更好的被实现了，所以说，在很大程度上，对于这个类的理解有助于我们更好的了解RocketMQ的核心内容。所以这个系列的详解会比较多一些，一行代码一行代码来分析这个类。

这个类来自于org.apache.rocketmq.client.impl.factory包是这个包下唯一的一个类。org.apache.rocketmq.client.impl.factory.MQClientInstance。那么下面就来进入到详细的分析中。

### 基础对象 ###

在之前的分析中，会看到很多的关于ConcurrentHashMap的使用，在MQClientInstance的属性中也定义了很多的ConcurrentHashMap，分别是

*  producerTable：用来缓存Group和生产者的对应关系
 *  consumerTable：用来缓存Group和消费者的对应关系
 *  adminExtTable：用来缓存Group和管理员的对应关系
 *  topicRouteTable：对Topic信息的缓存
 *  brokerAddrTable：对于集群中的Broker地址进行缓存
 *  brokerVersionTable：对于集群中Broker的版本信息进行缓存

private final ConcurrentMap<String/* group */, MQProducerInner> producerTable = new ConcurrentHashMap<String, MQProducerInner>();
    
    private final ConcurrentMap<String/* group */, MQConsumerInner> consumerTable = new ConcurrentHashMap<String, MQConsumerInner>();
    
    private final ConcurrentMap<String/* group */, MQAdminExtInner> adminExtTable = new ConcurrentHashMap<String, MQAdminExtInner>();
    
    private final ConcurrentMap<String/* Topic */, TopicRouteData> topicRouteTable = new ConcurrentHashMap<String, TopicRouteData>();
    
    private final ConcurrentMap<String/* Broker Name */, HashMap<Long/* brokerId */, String/* address */>> brokerAddrTable =
            new ConcurrentHashMap<String, HashMap<Long, String>>();
    private final ConcurrentMap<String/* Broker Name */, HashMap<String/* address */, Integer>> brokerVersionTable =
            new ConcurrentHashMap<String, HashMap<String, Integer>>();

从代码中会看到上面六个缓存表分别对应了六种数据类型。分别是

*  MQProducerInner：Producer的内置对象
 *  MQConsumerInner：Consumer的内置对象
 *  MQAdminExtInner：管理员的内置对象
 *  TopicRouteData：Topic分装信息
 *  <String/\* Broker Name */, HashMap<Long/* brokerId */, String/* address \*/> ：Broker的详细信息
 *  <String/\* Broker Name */, HashMap<String/* address \*/, Integer>> ：Broker详细版本信息

从上面的代码中可以看到，这里主要做了一个最简单的逻辑的封装，就是为这些对应关系找到如下的一种关系，它最终的目的就是为了实现，下面这种逻辑关系，而对于Broker的版本存储，也是为了更好的支持这种机制。在后续的分析中会对上面的这些数据提供一个详细的分析  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25paHVpMTIz_size_16_color_FFFFFF_t_70]  
  从上面图中可以看到，无论是消费者还是生产者，还是Broker NameServer，都是以集群的形式存在的，但是这里有一点需要说明的，无论是单点还是集群，这个集群内的机器标识都是可计数的。加之对集群进行命名，对于同一个集群内的机器数量都是可以计数的，所以说采用了ConcurrentHashMap来进行集群信息的维护。对于不同的集群无论这个集群有多大，最终都可以抽象为上面六种的数据类型。也就是说，在这个类中上面的内容都被抽象成了这个类中的对象，并且对这些对象进行的存储。在这个类中来控制这些对象之间的调用关系，以及为这些对象提供一些对应关系。也就是说它就是为了实现这个流程进行服务的对象，哪里有需要就在哪里New一个。看上去非常的孤独，没有任何的父类，也没有任何的子类。

### 构造方法 ###

在进一步了解之前首先来看看，这个对象的构造方法中都完成了那些操作，由于之前也分析了，这个对象就是了上面这个流程做服务的，所以说可以把它看做一个大集合，在这个集合里面通过构造函数进行初始化了很多的东西。下面的构造方法就为展示了很多的东西，例如ClientConfig 客户端配置类，instanceIndex实例索引，NettyConfig Netty的相关配置类，客户端的ClientRemotingProcessor 客户端远程执行，其中最为重要的一个就是mQClientAPIImpl的设置。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25paHVpMTIz_size_16_color_FFFFFF_t_70 1]

### 配置信息 ###

在分析，查看源码之后找到了如下的两个配置类，这个两个配置类一个是针对客户端的配置，一个是针对Netty的配置。那么分别来看看这两个配置类都有那些配置以及它的实现。也就是说在这个类中提供的最多的配置就是客户端的配置和Netty客户端的配置。这个两个配置就分别的堆RocketMQ作为客户端进行了封装另一个方面就是对作为Socket的Netty客户端的配置。从这两个角度上来看这个类就有了两个配置的所有的优点，所有的配置都是动态的配置。

private final ClientConfig clientConfig;
    private final NettyClientConfig nettyClientConfig;

### 加锁操作 ###

除了上面的配置以外还提供了下面两个东西，从功能的角度上就是保证了安全性，在集群的状态下，有多个实例需要进行管理就会涉及到多线程的问题，从它的命名角度上来看这两个锁似乎是为了支持基础功能，一个是NameServer上的注册表的获取，一个是心跳检测机制。先不讨论他们是什么，先来看看这两个对象使用的位置

private final Lock lockNamesrv = new ReentrantLock();
    private final Lock lockHeartbeat = new ReentrantLock();

**lockNamesrv**

public boolean updateTopicRouteInfoFromNameServer(final String topic, boolean isDefault,
            DefaultMQProducer defaultMQProducer)
    
    private void cleanOfflineBroker()

**lockHeartbeat**

public void sendHeartbeatToAllBrokerWithLock()
    
    
    private void unregisterClientWithLock(final String producerGroup, final String consumerGroup)

从上面四个使用锁的地方可以看到，分别是从远程获取Topic信息的时候和进行Broker心跳检测的时候，这两个时候由于会有多线程对当前信息进行读写，但是在同一时间只能有一个线程进行读写操作，所以这样的操作就需要进行加锁。对于这个重入锁的原理在后续的分析中会提到，这里只是对这个类进行详细的解读。所以这里先不对这个机制进行说明。

### 服务操作 ###

设置完配置，设置完锁操作之后，接下来做的事情就是一些服务的提供，这里提供了如下的四个服务

*  PullMessageService ：拉取消息服务
 *  RebalanceService ：负载均衡服务
 *  DefaultMQProducer : 默认的消息生产者，这个地方需要注意一下
 *  ConsumerStatsManager : 消费者状态管理。

private final PullMessageService pullMessageService;
     private final RebalanceService rebalanceService;
     private final DefaultMQProducer defaultMQProducer;
     private final ConsumerStatsManager consumerStatsManager;

上面的四个内容前三个都还很好理解，但是第四个有点难理解，这里我们看看在代码中这块内容是如何操作的

this.consumerStatsManager = new ConsumerStatsManager(this.scheduledExecutorService);

通过查看源码我们可以知道其实这个对象在这个类中并没有多大的使用，但是在org.apache.rocketmq.client.impl.consumer.DefaultMQPushConsumerImpl 类中有具体的使用，但是这里的使用也是作为了MQClientInstance类对象的调用，其实使用的还是这个对象。这里看到上面这个代码中其实传入了一个参数，这个参数表示的是如下的一个对象

private final ScheduledExecutorService scheduledExecutorService = Executors.newSingleThreadScheduledExecutor(new ThreadFactory() { 
            @Override
            public Thread newThread(Runnable r) { 
                return new Thread(r, "MQClientFactoryScheduledThread");
            }
        });

这里会知道ScheduledExecutorService 是一个线程池的操作，这个线程池是一个周期性执行的线程池，并且线程池中创建了一个线程对象，传入了一个参数。schedule 方法使用各种延迟创建任务，并返回一个可用于取消或检查执行的任务对象。scheduleAtFixedRate 和 scheduleWithFixedDelay 方法创建并执行某些在取消前一直定期运行的任务。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25paHVpMTIz_size_16_color_FFFFFF_t_70 2]  
  如图其实会看到在这个实例类中有很多的地方都使用了该对象，这里先不要去管这些方法都是干啥的，先来注意这些方法都准备干啥，只有知道了这些才能更好的知道这个MQClientInstance对象想要干啥，从而知道，整个的Client模块都有那些执行操作。

### 实际处理操作 ###

在很多的高级框架中把处理问题的实际逻辑都交给xxxProcessor 这样的对象或者是方法来执行，这个被称为处理器，也就是说Client实际上执行所有的处理操作都是通过这个对象来执行的。

private final ClientRemotingProcessor clientRemotingProcessor;
    
    this.clientRemotingProcessor = new ClientRemotingProcessor(this);
    
    this.mQClientAPIImpl = new MQClientAPIImpl(this.nettyClientConfig, this.clientRemotingProcessor, rpcHook, clientConfig);

会看到这个对象其实就是一个上面创建下面使用的的对象，并没有在全局的其他地方去使用到。

## 总结 ##

到这里，会知道其实MQClientInstance类说白了就是一个客户端的实现的转折点，这个类中封装了很多的基础操作就是为了支持上面图中所展示的操作，也就是说这个类就是上面操作在不同的场景中的埋点有了这个类作为基础，就是为后续的所有的操作打基础，如果还有对上面图中所展示的内容进行扩展的地方就需要基于这个类来进行扩展。在这个类的分析中也有留下的很多的问题和关键点，但是这些问题在后续的分享中还会不断的被提及到，并且这些分享也会加入更多的关于编程思想或者是设计模式的东西。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25paHVpMTIz_size_16_color_FFFFFF_t_70]: /images/20230528/c544884fe49045edbf2b01fa913152df.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25paHVpMTIz_size_16_color_FFFFFF_t_70 1]: /images/20230528/900cc2c161434c209b1b4557abec8079.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25paHVpMTIz_size_16_color_FFFFFF_t_70 2]: /images/20230528/308315295ccc4800bc9201361658598f.png