MQ底層類MQClientInstance和ProcessQueue分析

本篇內(nèi)容介紹了“MQ底層類MQClientInstance和ProcessQueue分析”的有關(guān)知識(shí)，在實(shí)際案例的操作過程中，不少人都會(huì)遇到這樣的困境，接下來就讓小編帶領(lǐng)大家學(xué)習(xí)一下如何處理這些情況吧！希望大家仔細(xì)閱讀，能夠?qū)W有所成！

基石MQClientInstance

DefaultMQProducerImpl 、DefaultMQPullConsumerImpl、DefaultMQPushConsumerImpl  三個(gè)對(duì)象的成員變量都包含 同一個(gè)成員對(duì)象：

private MQClientInstance mQClientFactory;

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

MQClientInstance功能如下：

• MQClientInstance 負(fù)責(zé)與NameServer 通信進(jìn)行心跳維持、根據(jù)Topic獲取Broker地址；

• MQClientInstance負(fù)責(zé)與Broker通信進(jìn)行收發(fā)消息、ReBalance；

普通情況下，一個(gè)用到RocketMQ的生產(chǎn)端或者消費(fèi)端，或者說一個(gè)JVM進(jìn)程只要有一個(gè)MQClientInstance實(shí)例就夠了.如果不手動(dòng)通過setInstanceName設(shè)置，這個(gè)時(shí)候創(chuàng)建的一個(gè)或者多個(gè)Consumer或者Producer底層使用的都是同一個(gè)MQClientInstance實(shí)例(默認(rèn)名字是"DEFAULT").

但是有些情況下只有一個(gè)MQClientInstance對(duì)象是不夠的,比如一個(gè)Java程序需要連接兩個(gè)MQ集群的，從一個(gè)集群消費(fèi)，發(fā)送到另外一個(gè)集群，一個(gè)MQClientInstance對(duì)象則無法支持這種場(chǎng)景,此時(shí)一定要手動(dòng)指定不同的InstanceName,底層就會(huì)創(chuàng)建兩個(gè)MQClientInstance對(duì)象。

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的生成規(guī)則：

如果同一臺(tái)機(jī)器上有多個(gè)服務(wù)集成了MQ的客戶端，進(jìn)程ID

MessageQueue在client端的快照類ProcessQueue

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

“MQ底層類MQClientInstance和ProcessQueue分析”的內(nèi)容就介紹到這里了，感謝大家的閱讀。如果想了解更多行業(yè)相關(guān)的知識(shí)可以關(guān)注億速云網(wǎng)站，小編將為大家輸出更多高質(zhì)量的實(shí)用文章！