Spring Integration MQTT共享订阅单独失效，搭配普通订阅致重复消费求助

问题背景 #

使用版本：

• spring-integration-mqtt: 6.4.3
• org.eclipse.paho.mqttv5.client: 1.2.5
• spring-boot: 3.4.4

问题表现：
单独配置MQTT共享订阅（格式$share/group/{topic}）时无法接收消息，必须同时订阅普通主题和共享主题才能生效，但此时每条消息会被消费两次。

原因分析 #

1. 客户端配置缺失MQTTv5特性：若客户端管理器未正确配置MQTTv5连接参数，Broker可能无法识别共享订阅格式的主题。
2. Broker兼容性问题：部分MQTT Broker默认未开启共享订阅支持，或需特定配置才能解析$share前缀的主题。
3. 日志缺失导致排查盲区：未启用调试日志时，无法发现订阅失败的具体原因（如权限不足、主题格式错误等）。

解决方案 #

1. 完善MQTTv5客户端管理器配置 #

确保客户端管理器正确配置MQTTv5协议参数，让Broker识别MQTTv5连接：

@Bean
public ClientManager<IMqttAsyncClient, MqttConnectionOptions> mqttClientManager() {
    MqttConnectionOptions connectionOptions = new MqttConnectionOptions();
    // 启用MQTTv5 clean start（按需调整）
    connectionOptions.setCleanStart(true);
    // 配置Broker地址、认证信息
    connectionOptions.setServerURIs(new String[]{"tcp://your-mqtt-broker:1883"});
    connectionOptions.setUserName("mqtt-username");
    connectionOptions.setPassword("mqtt-password".getBytes());
    
    // 多实例部署时，客户端ID需唯一，避免Broker断开重复连接
    return new Mqttv5ClientManager(connectionOptions, "mqtt-client-" + UUID.randomUUID());
}

2. 正确初始化共享订阅适配器 #

直接使用共享主题构造适配器，无需添加普通主题：

@Bean
public List<IntegrationFlow> autoRegisterProcessors(
        ClientManager<IMqttAsyncClient, MqttConnectionOptions> clientManager,
        List<MqttMessageProcessor> processors,
        IntegrationFlowContext flowContext) {

    return processors.stream()
            .map(processor -> {
                String topic = processor.getTopic();
                String sharedTopic = "$share/group/" + topic;
                
                // 直接用共享主题初始化适配器
                Mqttv5PahoMessageDrivenChannelAdapter adapter = 
                        new Mqttv5PahoMessageDrivenChannelAdapter(clientManager, sharedTopic);
                adapter.setQos(processor.getQos()); // 设置对应QoS等级
                
                IntegrationFlow flow = IntegrationFlow
                        .from(adapter)
                        .handle(processor::process)
                        .get();
                
                flowContext.registration(flow).register();
                return flow;
            })
            .toList();
}

3. 验证Broker共享订阅支持 #

• Mosquitto：确保版本≥2.0，默认启用共享订阅，无需额外配置。
• EMQX：默认支持共享订阅，直接使用即可。
• 发送测试消息到普通主题（如topic），检查共享订阅是否能正常接收消息。

4. 启用调试日志排查细节 #

在application.yml中添加日志配置，查看订阅过程的详细信息：

logging:
  level:
    org.springframework.integration.mqtt: DEBUG
    org.eclipse.paho.mqttv5.client: DEBUG

重点关注日志中是否有Subscribed to topic: $share/group/xxx的成功提示，若存在订阅失败日志，根据错误信息修正（如权限配置、Broker连接问题）。

注意事项 #

• 多实例部署时，客户端ID必须唯一，否则Broker会断开重复连接的客户端。
• 若仍出现重复消费，需检查是否存在重复订阅逻辑（如同时订阅普通主题和共享主题）。

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