如何在Spring Integration MQTT中实现RabbitMQ共享订阅负载均衡？

要解决消息被单一实例接收的问题，核心是利用RabbitMQ MQTT插件的共享订阅机制，结合Spring Integration的正确配置，实现消息在同组客户端间的均衡分配，具体步骤如下：

1. 确保RabbitMQ支持共享订阅 #

首先确认RabbitMQ已启用MQTT插件，执行以下命令启用：

rabbitmq-plugins enable rabbitmq_mqtt

RabbitMQ 3.8及以上版本的MQTT插件默认支持共享订阅，无需额外配置。

2. 配置唯一的MQTT客户端ID #

每个应用实例必须使用唯一的clientId，否则RabbitMQ会踢掉重复ID的连接，导致只有一个实例保持在线。可以结合端口号、实例ID等动态生成clientId：

Java配置示例 #

@Bean
public MqttPahoClientFactory mqttClientFactory() {
    DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
    MqttConnectOptions connectOptions = new MqttConnectOptions();
    connectOptions.setServerURIs(new String[]{"tcp://rabbitmq-host:1883"});
    connectOptions.setUserName("mqtt-user");
    connectOptions.setPassword("mqtt-password".toCharArray());
    // 用端口号生成唯一clientId
    connectOptions.setClientId("mqtt-consumer-" + System.getProperty("server.port"));
    factory.setConnectionOptions(connectOptions);
    return factory;
}

YAML配置示例 #

spring:
  mqtt:
    client:
      url: tcp://rabbitmq-host:1883
      username: mqtt-user
      password: mqtt-password
      client-id: mqtt-consumer-${server.port} # 动态绑定端口生成唯一ID

3. 使用共享订阅主题格式 #

RabbitMQ的共享订阅需要使用特定的主题前缀：$share/{group-name}/<your-topic>，其中：

• group-name是消费组标识，同一组内的实例必须使用相同的group-name
• <your-topic>是实际要订阅的主题（比如你的example_topic）

Spring Integration订阅配置示例 #

Java配置

@Bean
public MessageProducerSupport mqttInboundAdapter(MqttPahoClientFactory clientFactory) {
    // 使用共享订阅主题前缀
    String sharedTopic = "$share/my-consumer-group/example_topic";
    MqttPahoMessageDrivenChannelAdapter adapter = new MqttPahoMessageDrivenChannelAdapter(
            clientFactory, sharedTopic);
    adapter.setQos(1); // 根据业务需求设置QoS级别
    adapter.setOutputChannel(mqttInputChannel()); // 绑定消息处理通道
    adapter.setConverter(new DefaultPahoMessageConverter());
    return adapter;
}

@Bean
public MessageChannel mqttInputChannel() {
    return new DirectChannel();
}

// 消息处理逻辑
@ServiceActivator(inputChannel = "mqttInputChannel")
public void handleMqttMessage(String payload) {
    System.out.println("Received message: " + payload + " on instance port: " + System.getProperty("server.port"));
}

YAML配置（配合Spring Boot自动配置）

spring:
  mqtt:
    inbound:
      topics: $share/my-consumer-group/example_topic # 共享订阅主题
      qos: 1

4. 验证效果 #

启动两个端口分别为8081、8082的实例，用MQTTX向example_topic发送消息，此时RabbitMQ会将消息均衡分配给两个实例，不会再出现全部消息被单一实例接收的情况。

内容的提问来源于stack exchange，提问作者Romillion