基于Spring Integration MQTT实现同一连接下的发布与订阅

当然可以做到！你当前代码里出现发布时原有订阅连接断开的问题，核心原因是发布者和订阅者使用了不同的Client ID——MQTT协议要求每个连接的Client ID必须唯一，所以当你的发布者以siSamplePublisher建立连接时，Broker会强制断开订阅者siSampleConsumer的旧连接，反之亦然，这就导致了连接反复断开重连的现象。

解决方案：共享同一Client ID与连接 #

只需要让发布者和订阅者使用相同的Client ID，Spring Integration的Paho组件会自动复用同一个MQTT客户端连接，这样就能在同一连接上同时进行消息发布与订阅，不会再出现互相踢掉连接的问题。

下面是修改后的完整示例代码：

@Bean
public MqttPahoClientFactory mqttClientFactory() {
    DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
    factory.setServerURIs("tcp://localhost:1883");
    factory.setUserName("guest");
    factory.setPassword("guest");
    return factory;
}

// 统一配置MQTT连接参数（可选，但推荐集中管理）
@Bean
public MqttConnectOptions mqttConnectOptions() {
    MqttConnectOptions options = new MqttConnectOptions();
    options.setUserName("guest");
    options.setPassword("guest".toCharArray());
    options.setServerURIs(new String[]{"tcp://localhost:1883"});
    options.setCleanSession(false); // 若需要持久化订阅，设置为false
    options.setKeepAliveInterval(60); // 心跳间隔，维持连接
    return options;
}

// 发布者：使用统一的Client ID
@Bean
public IntegrationFlow mqttOutFlow() {
    return IntegrationFlows.from(CharacterStreamReadingMessageSource.stdin(), 
                                 e -> e.poller(Pollers.fixedDelay(1000)))
            .transform(p -> p + " sent to MQTT")
            .handle(mqttOutbound())
            .get();
}

@Bean
public MessageHandler mqttOutbound() {
    // 关键：使用和订阅者相同的Client ID
    MqttPahoMessageHandler messageHandler = new MqttPahoMessageHandler("siSampleSharedClient", mqttClientFactory());
    messageHandler.setAsync(true);
    messageHandler.setDefaultTopic("siSampleTopic");
    messageHandler.setConnectOptions(mqttConnectOptions());
    // 可选：设置发布回调，监听发布是否成功
    messageHandler.setCallback(new MqttCallback() {
        @Override
        public void connectionLost(Throwable cause) {
            // 处理连接丢失逻辑
        }

        @Override
        public void messageArrived(String topic, MqttMessage message) throws Exception {
            // 订阅消息的回调，发布者主要用下面的deliveryComplete
        }

        @Override
        public void deliveryComplete(IMqttDeliveryToken token) {
            // 发布完成的ACK回调，可用于确认消息已发送
            System.out.println("Message delivered: " + token.getMessageId());
        }
    });
    return messageHandler;
}

// 消费者：使用统一的Client ID
@Bean
public IntegrationFlow mqttInFlow() {
    return IntegrationFlows.from(mqttInbound())
            .transform(p -> p + ", received from MQTT")
            .handle(logger())
            .get();
}

private LoggingHandler logger() {
    LoggingHandler loggingHandler = new LoggingHandler("INFO");
    loggingHandler.setLoggerName("siSample");
    return loggingHandler;
}

@Bean
public MessageProducerSupport mqttInbound() {
    // 关键：使用和发布者相同的Client ID
    MqttPahoMessageDrivenChannelAdapter adapter = new MqttPahoMessageDrivenChannelAdapter("siSampleSharedClient", 
                                                                                           mqttClientFactory(), 
                                                                                           "siSampleTopic");
    adapter.setCompletionTimeout(5000);
    adapter.setConverter(new DefaultPahoMessageConverter());
    adapter.setQos(1);
    adapter.setConnectOptions(mqttConnectOptions());
    return adapter;
}

关键说明 #

1. 共享Client ID：发布者和订阅者的Client ID统一为siSampleSharedClient，Broker只会维护一个该ID的连接，发布和订阅操作都在这个连接上完成。
2. 连接参数集中管理：通过MqttConnectOptions统一配置连接的用户名、密码、心跳等参数，避免重复配置。
3. 请求-响应场景优化：当需要等待发布后的响应时，同一连接可以直接订阅响应主题，无需处理两个独立连接的同步问题，代码逻辑会更简洁。
4. 回调与可靠性：可以通过setCallback监听发布的ACK确认，确保消息成功发送；设置cleanSession=false可以让Broker持久化订阅关系，客户端重连后不会丢失未接收的消息。

这样修改后，你的发布和订阅操作就会共用同一个MQTT连接，不会再出现连接频繁断开重连的问题，同时也能更顺畅地处理需要等待响应的业务场景。

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