告别重复代码！RocketMQ多实例Starter实战：开发到开源发布一步到位

烟雨平生

发布于 2026-04-14 18:34:02

330

用过RocketMQ的都知道，官网只提供了一个集群的starter，如果业务域进行消息隔离，使用了不同的集群，就只能copy、paste这种样板式代码，好烦！！！

最近闲来有精神，就写一个支持多集群的声明式的RocketMQ spring-boot-starter组件，来实现代码的代码的共享。

https://central.sonatype.com/artifact/io.github.helloworldtang/rocketmq-multi-instance-spring-boot-starter

在微服务架构中，RocketMQ 作为高性能的消息中间件被广泛使用。官方提供的 rocketmq-spring-boot-starter 虽然开箱即用，但在面对复杂场景时仍显不足。例如，当一个应用需要同时连接多个 RocketMQ 集群，或同一集群的不同 NameServer 时，官方 Starter 的配置往往变得冗长且难以维护。

为此，我设计并实现了一个支持多实例配置的 Spring Boot Starter，不仅能够优雅管理多个 RocketMQ 连接，还将其发布至 Maven 中央仓库，供开发者直接引入使用。

<dependency>
    <groupId>io.github.helloworldtang</groupId>
    <artifactId>rocketmq-multi-instance-spring-boot-starter</artifactId>
    <version>0.1.0</version>
</dependency>

本文将完整分享从架构设计、核心代码实现，到最终发布上线的全过程。

一、架构设计与实现思路

1. 痛点与设计目标

多实例隔离：通过配置文件灵活定义多个 RocketMQ 实例（包括 NameServer、AccessKey 等配置）。
声明式消费：保留类似 @RocketMQMessageListener 的简洁注解风格，同时支持指定消费实例。
灵活的生产方式：支持 Spring 容器内依赖注入，也提供静态方法便于在工具类等非托管 Bean 中获取 Producer。

2. 核心实现解析

动态注册 Producer 借助 Spring 的 ImportBeanDefinitionRegistrar，我们可以在启动阶段动态读取配置，为每个实例注册对应的 DefaultMQProducer Bean。

// 遍历配置，动态注册 Producer Bean
instances.forEach((name, config) -> {
    String beanName = name + "Producer";
    if (!registry.containsBeanDefinition(beanName)) {
        BeanDefinitionBuilder builder = 
            BeanDefinitionBuilder.genericBeanDefinition(DefaultMQProducer.class);
        builder.addConstructorArgValue(name + "-producer-group");
        builder.addPropertyValue("namesrvAddr", config.getNameServer());
        // 设置其他参数...
        registry.registerBeanDefinition(beanName, builder.getBeanDefinition());
    }
});

同时，为了在任意位置便捷获取 Producer，设计了一个容器类 RocketMQProducerContainer，提供静态获取方法：

public static DefaultMQProducer get(String instanceId) {
    // 从内部容器返回对应实例
}

声明式 Consumer
自定义注解 @RocketMQListener，增加 instanceId 属性以指定消息来源：

@Target(ElementType.TYPE)
@Retention(RetentionPolicy.RUNTIME)
@Component
public @interface RocketMQListener {
    String instanceId();      // 指定 RocketMQ 实例
    String topic();
    String consumerGroupId();
    // 其他配置...
}

通过实现 SmartInitializingSingleton 接口，在 Bean 初始化完成后扫描该注解，并启动对应的 RocketMQ 消费者。

二、发布至 Maven 中央仓库

实现代码只是第一步，将其共享给社区才真正创造价值。Sonatype 提供了新的 Central Portal，显著简化了发布流程。

1. 前期准备

注册账号：使用 GitHub 账号登录 central.sonatype.com。
GPG 签名：所有发布到中央仓库的组件都必须经过签名。

# 生成密钥对
gpg --gen-key

# 上传公钥至服务器（关键步骤，否则验证会失败）
gpg --keyserver keys.openpgp.org --send-keys <你的KeyID>

2. 配置 pom.xml

中央仓库对项目元数据有严格要求，需准确配置以下信息：

项目名称、描述与 URL
开源协议（licenses）
开发者信息（developers）
源码仓库信息（scm）

3. 使用新版发布插件 在 pom.xml 的 build 环节引入 central-publishing-maven-plugin，实现自动上传与发布：

<plugin>
    <groupId>org.sonatype.central</groupId>
    <artifactId>central-publishing-maven-plugin</artifactId>
    <version>0.5.0</version>
    <extensions>true</extensions>
    <configuration>
        <publishingServerId>central</publishingServerId>
        <autoPublish>true</autoPublish>  <!-- 自动发布，无需手动操作 -->
    </configuration>
</plugin>

4. 配置 settings.xml

在 Maven 的 settings.xml 中添加 Sonatype 的 Token 认证信息：

<server>
    <id>central</id>
    <username>你的Token用户名</username>
    <password>你的Token密码</password>
</server>

完成配置后，执行 mvn deploy

mvn -s /path/to/settings.xml \
    -Dmaven.repo.local=~/.m2/repository \
    -Prelease -DskipTests \
    -pl rocketmq-multi-instance-spring-boot-starter -am deploy

稍等片刻，你的组件便会出现在 Maven 中央仓库中。

三、快速上手指南

只需三步，即可在项目中引入并使用该 Starter。

Step 1：引入依赖

<dependency>
    <groupId>io.github.helloworldtang</groupId>
    <artifactId>rocketmq-multi-instance-spring-boot-starter</artifactId>
    <version>0.1.0</version>
</dependency>

Step 2：配置多实例
在 application.yml 中定义多个 RocketMQ 集群：

rocketmq:
  instances:
    bj-cluster:                # 北京集群,唯一标识，可以用来获取DefaultMQProducer
      name-server: 192.168.1.10:9876
    sh-cluster:                # 上海集群,唯一标识，可以用来获取DefaultMQProducer
      name-server: 192.168.2.20:9876

Step 3：发送与消费消息

发送消息（通过静态方法）：

DefaultMQProducer producer = RocketMQProducerContainer.get("bj-cluster");//bj-cluster为yml配置的集群ID
if (producer != null) {
    producer.send(new Message("TopicTest", "TagA", "Hello Beijing".getBytes()));
}

消费消息（注解驱动）：

@Component
@RocketMQListener(
    instanceId = "sh-cluster",       // 指定消费上海集群
    topic = "TopicTest",
    consumerGroupId = "sh-consumer-group"
)
public class MyConsumer implements RocketMQMessageListener {
    @Override
    public void onMessage(MessageExt message) {
        System.out.println("收到上海集群的消息: " + new String(message.getBody()));
    }
}