如何在同一個(gè)Java進(jìn)程中連接多個(gè)RocketMQ服務(wù)器

前言

我們都知道，RocketMQ在代碼級(jí)別對(duì)連接服務(wù)器進(jìn)行了限制，基本上可以理解為一個(gè)JVM進(jìn)程中只能連接一個(gè)NameServer，但實(shí)際應(yīng)用場(chǎng)景中，我們可能會(huì)在架構(gòu)設(shè)計(jì)層面上對(duì)RocketMQ進(jìn)行了職能上的劃分，規(guī)定了A服務(wù)處理A類消息，而B服務(wù)處理B類消息，這時(shí)我們應(yīng)該如何解決這個(gè)問題呢？

問題的根源

我們從代碼層級(jí)來分析到底為什么會(huì)產(chǎn)生“一個(gè)JVM實(shí)例只能連接一個(gè)NameServer”。
RocketMQ Client有一個(gè)核心類MQClientManager,在我們需要使用MQ Client實(shí)例的時(shí)候，實(shí)際上都是通過它的getAndCreateMQClientInstance方法進(jìn)行創(chuàng)建的；名稱比較拗口，同時(shí)是Get和Create，這不太符合我們所說的設(shè)計(jì)單一性原則，但這不是我們討論的重點(diǎn)，我們看一看這個(gè)方法的實(shí)現(xiàn)

    public MQClientInstance getAndCreateMQClientInstance(ClientConfig clientConfig, RPCHook rpcHook) {
        String clientId = clientConfig.buildMQClientId();
        MQClientInstance instance = (MQClientInstance)this.factoryTable.get(clientId);
        if (null == instance) {
            instance = new MQClientInstance(clientConfig.cloneClientConfig(), this.factoryIndexGenerator.getAndIncrement(), clientId, rpcHook);
            MQClientInstance prev = (MQClientInstance)this.factoryTable.putIfAbsent(clientId, instance);
            if (prev != null) {
                instance = prev;
                log.warn("Returned Previous MQClientInstance for clientId:[{}]", clientId);
            } else {
                log.info("Created new MQClientInstance for clientId:[{}]", clientId);
            }
        }

        return instance;
    }

代碼不復(fù)雜，我們可以看到它利用客戶的配置信息生成一個(gè)固定的clientId，以此去緩存factoryTable中查找，不存在才會(huì)創(chuàng)建全新一個(gè)實(shí)例。
那么，可以理解一個(gè)clientID僅能存在一個(gè)連接實(shí)例了，可這個(gè)clientId是怎么產(chǎn)生的呢？繼續(xù)跟蹤看看這段代碼

    public String buildMQClientId() {
        StringBuilder sb = new StringBuilder();
        sb.append(this.getClientIP());
        sb.append("@");
        sb.append(this.getInstanceName());
        if (!UtilAll.isBlank(this.unitName)) {
            sb.append("@");
            sb.append(this.unitName);
        }

        return sb.toString();
    }

代碼層面上對(duì)clientId進(jìn)行了約定，格式為“ClientIp@InstanceName”格式，當(dāng)unitName不為空的時(shí)候還會(huì)在后面加上“@unitName”。

怎么解決？

從代碼分析上我們可以知道，為了創(chuàng)建多實(shí)例，我們可以

1. 設(shè)定不同的instanceName：

   instanceName從哪來的？

   instanceName = System.getProperty("rocketmq.client.name", "DEFAULT");

   從系統(tǒng)屬性中讀取出來的，也就是一般在JVM啟動(dòng)時(shí)設(shè)定的。。。
   可以變嗎？當(dāng)然，你可以通過代碼去做到，但這么做的話，你會(huì)失去讓人理解你代碼的能力的，哈哈
   這就是為什么多少RocketMQ Client都只能連接一個(gè)服務(wù)器的原因了，它根本不考慮服務(wù)器是誰(shuí)，僅關(guān)心自己，自私的家伙！

2. 設(shè)定不同的unitName

除此之外還有其它解決方案嗎？我仔細(xì)從網(wǎng)絡(luò)上翻了一輪，沒看到什么好方法，是大家都沒這個(gè)場(chǎng)景還是有其它好辦法解決了呢？歡迎大家討論~

方法3

在上一篇博文來自平行世界的救贖里面，我做了個(gè)工具sandbox，我提供的方法3就是依托于這個(gè)工具。
sandbox通過代碼隔離的方式，將另一份類定義放入沙箱中運(yùn)行，從而實(shí)現(xiàn)多個(gè)實(shí)例完全隔離的效果。
MQClientManager通過緩存方式，以clientId作為key值存儲(chǔ)到自身實(shí)例當(dāng)中，為了實(shí)現(xiàn)多個(gè)Client，那么前兩種方法的邏輯是修改clientId實(shí)現(xiàn)多個(gè)實(shí)例，而方法3的邏輯則是“既然你的緩存已經(jīng)有這個(gè)key，我就換個(gè)緩存”，本質(zhì)就是“你這個(gè)鍋不裝我，我就換個(gè)鍋”。

怎么做？

這里我使用一個(gè)springboot項(xiàng)目作為演示案例。
通過springboot的Configuration將多個(gè)RocketMQ Client進(jìn)行注冊(cè)，再定義一個(gè)Controller接收不同請(qǐng)求去發(fā)送MQ消息，最后加上啟動(dòng)類。

我們先從pom文件中引入包（我沒有推上maven倉(cāng)庫(kù)，各位可以從github/gitee上下載），代碼如下

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <groupId>me.van</groupId>
    <artifactId>rocket-mq-multi-client-test</artifactId>
    <version>1.0-SNAPSHOT</version>
    <packaging>jar</packaging>

    <name>測(cè)試多個(gè)rocketmq client共存</name>

    <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>2.1.6.RELEASE</version>
    </parent>

    <properties>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
        <start-class>me.van.App</start-class>
        <java.version>1.8</java.version>
        <lombok.version>1.14.8</lombok.version>
    </properties>

    <dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>
        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
            <version>${lombok.version}</version>
            <scope>provided</scope>
        </dependency>

        <dependency>
            <groupId>org.apache.rocketmq</groupId>
            <artifactId>rocketmq-client</artifactId>
            <version>4.4.0</version>
        </dependency>
        <dependency>
            <groupId>me.van</groupId>
            <artifactId>sandbox</artifactId>
            <version>1.0.0</version>
        </dependency>
    </dependencies>
    <build>
        <plugins>
            <plugin>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-maven-plugin</artifactId>
            </plugin>
        </plugins>
    </build>

</project>

此處引入了apache的rocketmq-client組件作為mq客戶端，也就是存在前面所說的問題的組件。

啟動(dòng)類

package me.van;

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;

@SpringBootApplication
public class App {
    public static void main(String[] args) {
        SpringApplication.run(App.class, args);
    }
}

非常的簡(jiǎn)單，沒什么好介紹的。

配置類

package me.van;

import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.MQProducer;
import org.springframework.beans.factory.annotation.Autowire;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class AppConfig {

    @Bean(autowire = Autowire.BY_NAME, value = "producer")
    MQProducer producer() throws MQClientException {
        DefaultMQProducer producer = new DefaultMQProducer();
        initProducer(producer, "a.io:9876;b.io:9876");
        return producer;
    }

    @Bean(autowire = Autowire.BY_NAME, value = "producer_sandbox1")
    MQProducer producerSandbox1() throws MQClientException, SandboxCannotCreateObjectException {
        DefaultMQProducer producer = createProducerInSandbox();
        initProducer(producer, "x.io:9876;y.io:9876");
        return producer;
    }

    @Bean(autowire = Autowire.BY_NAME, value = "producer_sandbox2")
    MQProducer producerSandbox2() throws MQClientException, SandboxCannotCreateObjectException {
        DefaultMQProducer producer = createProducerInSandbox();
        initProducer(producer, "1.io:9876;2.io:9876");
        return producer;
    }

    private DefaultMQProducer createProducerInSandbox() throws SandboxCannotCreateObjectException {
        Sandbox sandbox = new Sandbox("org.apache.rocketmq.client");
        return sandbox.createObject(DefaultMQProducer.class);
    }

    private void initProducer(DefaultMQProducer producer, String namesrvAddr) throws MQClientException {
        producer.setNamesrvAddr(namesrvAddr);
        producer.setProducerGroup("test-group");
        producer.setRetryAnotherBrokerWhenNotStoreOK(true);
        producer.start();
    }
}

這里可以看到，producer對(duì)象是直接new 出來的DefaultMQProducer，而producer_sandbox1和producer_sandbox2是通過不同的沙箱創(chuàng)建出來的；三個(gè)client分別連接到不同的NameServer中，同時(shí)其它屬性保持一致。

Controller

package me.van;

import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.MQProducer;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.exception.RemotingException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;

@RestController
public class TestController {
    @Autowired
    MQProducer producer;
    @Autowired
    MQProducer producer_sandbox1;
    @Autowired
    MQProducer producer_sandbox2;

    @GetMapping("/")
    public String hello(){
        return "hello world";
    }

    @GetMapping("/send")
    public String send(String msg){
        if(null == msg) return "msg is null";

        String returnMsg = "";
        Message message = new Message("topic-test-multi-mq-client", msg.getBytes());
        try {
            producer.send(message);
            returnMsg += "原生producer發(fā)送完成<br/>";

            producer_sandbox1.send(message);
            returnMsg += "第一個(gè)沙箱內(nèi)producer發(fā)送完成<br/>";

            producer_sandbox2.send(message);
            returnMsg += "第二個(gè)沙箱內(nèi)producer發(fā)送完成<br/>";
        } catch (MQClientException | InterruptedException | RemotingException | MQBrokerException e) {
            returnMsg += "發(fā)送過程出現(xiàn)異常：" + e.getMessage();
        }
        return returnMsg;
    }
}

通過send方法同時(shí)向三個(gè)producer發(fā)送消息。

測(cè)試一下

運(yùn)行App，等幾秒鐘啟動(dòng)完畢，訪問http://localhost:8080/send，返回

msg is null

訪問，http://localhost:8080/send?msg=test

代碼地址

github: https://github.com/vancoo/multi-mq-demo
gitee: https://gitee.com/vancoo/multi-mq-demo

參考文檔： 來自平行世界的救贖