MQ底层类MQClientInstance和ProcessQueue分析

本篇内容介绍了“MQ底层类MQClientInstance和ProcessQueue分析”的有关知识，在实际案例的操作过程中，不少人都会遇到这样的困境，接下来就让小编带领大家学习一下如何处理这些情况吧！希望大家仔细阅读，能够学有所成！

基石MQClientInstance

DefaultMQProducerImpl 、DefaultMQPullConsumerImpl、DefaultMQPushConsumerImpl  三个对象的成员变量都包含 同一个成员对象：

private MQClientInstance mQClientFactory;

this.mQClientFactory = MQClientManager.getInstance().getAndCreateMQClientInstance(this.defaultMQPushConsumer, this.rpcHook);

MQClientInstance功能如下：

• MQClientInstance 负责与NameServer 通信进行心跳维持、根据Topic获取Broker地址；

• MQClientInstance负责与Broker通信进行收发消息、ReBalance；

普通情况下，一个用到RocketMQ的生产端或者消费端，或者说一个JVM进程只要有一个MQClientInstance实例就够了.如果不手动通过setInstanceName设置，这个时候创建的一个或者多个Consumer或者Producer底层使用的都是同一个MQClientInstance实例(默认名字是"DEFAULT").

但是有些情况下只有一个MQClientInstance对象是不够的,比如一个Java程序需要连接两个MQ集群的，从一个集群消费，发送到另外一个集群，一个MQClientInstance对象则无法支持这种场景,此时一定要手动指定不同的InstanceName,底层就会创建两个MQClientInstance对象。

public class MQClientManager {
  
    private static MQClientManager instance = new MQClientManager();
    private AtomicInteger factoryIndexGenerator = new AtomicInteger();
    private ConcurrentMap<String/* clientId */, MQClientInstance> factoryTable =
        new ConcurrentHashMap<String, MQClientInstance>();

ClientId的生成规则：

如果同一台机器上有多个服务集成了MQ的客户端，进程ID

MessageQueue在client端的快照类ProcessQueue

路径：org.apache.rocketmq.client.impl.consumer

“MQ底层类MQClientInstance和ProcessQueue分析”的内容就介绍到这里了，感谢大家的阅读。如果想了解更多行业相关的知识可以关注亿速云网站，小编将为大家输出更多高质量的实用文章！