Spring Integration MQTTv5重连后无法自动重新订阅主题求助

环境版本 #

• Spring Boot 3.0.2
• Spring Integration MQTT 6.0.2

问题现象 #

配置了自动重连的MQTTv5客户端，在断开连接并恢复后，无法自动重新订阅主题，必须重启应用才能接收消息。

现有代码实现 #

配置类 #

@Configuration
class MnpConfiguration {
    
    @Bean
    fun mqttInputChannel(): MessageChannel {
        return DirectChannel()
    }

    @Bean
    fun clientManager(): ClientManager<IMqttAsyncClient, MqttConnectionOptions> {
        val connectionOptions = MqttConnectionOptions()
        connectionOptions.serverURIs = arrayOf("mqtt://localhost:1883")
        connectionOptions.connectionTimeout = 3000
        connectionOptions.maxReconnectDelay = 1000
        connectionOptions.isAutomaticReconnect = true
      
        val clientManager = Mqttv5ClientManager(connectionOptions, UUID.randomUUID().toString())
        return clientManager
    }

    @Bean
    fun inbound(): MessageProducer {
        val manager = clientManager()

        val adapter = Mqttv5PahoMessageDrivenChannelAdapter(
            manager,
            "TEST"
        )

        adapter.setCompletionTimeout(1000)
        adapter.setPayloadType(String::class.java)
        adapter.setQos(0)
        adapter.outputChannel = mqttInputChannel()

        return adapter
    }
}

消息处理类 #

@Component
class MnpStream {

    @Bean
    @ServiceActivator(inputChannel = "mqttInputChannel")
    fun handleMnpLicense() = MessageHandler {
        println(String(it.payload as ByteArray))
    }
}

临时解决方法 #

添加自定义回调手动触发订阅恢复：

manager.addCallback {
   if (it && !adapter.connectionInfo.isCleanStart) {
       adapter.connectComplete(false)
   }
}

正式解决方案 #

方案1：基于MQTT会话持久化自动恢复订阅 #

这是MQTT协议层面的标准解决方案：

1. 在MqttConnectionOptions中设置cleanStart = false，告知Broker保留客户端会话：

connectionOptions.isCleanStart = false

2. 使用固定的客户端ID，避免每次启动生成新UUID，确保Broker能关联到之前的会话：

val clientManager = Mqttv5ClientManager(connectionOptions, "your-fixed-client-id")

客户端重连时，Broker会自动恢复该客户端的订阅关系，无需额外代码干预。

方案2：利用Spring Integration内置机制自动恢复 #

Spring Integration MQTT 6.0+版本的Mqttv5PahoMessageDrivenChannelAdapter已内置重连后订阅恢复逻辑，只需确保：

• 客户端ID固定
• cleanStart设为false
• 移除自定义的重连回调，避免干扰内置逻辑

方案3：自定义重连订阅逻辑（适用于动态客户端ID场景） #

如果业务必须使用动态客户端ID，可在重连完成后手动触发订阅：

@Bean
fun inbound(): MessageProducer {
    val manager = clientManager()
    val adapter = Mqttv5PahoMessageDrivenChannelAdapter(
        manager,
        "TEST"
    )
    
    adapter.setCompletionTimeout(1000)
    adapter.setPayloadType(String::class.java)
    adapter.setQos(0)
    adapter.outputChannel = mqttInputChannel()

    // 注册重连回调，触发重新订阅
    manager.addCallback { isReconnect ->
        if (isReconnect) {
            adapter.subscribe()
        }
    }

    return adapter
}

此方式需注意避免重复订阅，可通过判断适配器状态优化逻辑。

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