使用 Spring Integration MQTT 以相同的连接发布和订阅

Question

由于 MQTT 的设计，您只能使用唯一的客户端 ID 建立连接，是否可以使用相同的连接在 Spring Framework/Boot 中使用集成进行发布和订阅？

以这个非常简单的例子，它会连接到MQTT代理来订阅和获取消息，但是如果你想发布一条消息，第一个连接会在消息发送后断开并重新连接。

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

// publisher

@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() {
    MqttPahoMessageHandler messageHandler = new MqttPahoMessageHandler("siSamplePublisher", mqttClientFactory());
    messageHandler.setAsync(true);
    messageHandler.setDefaultTopic("siSampleTopic");
    return messageHandler;
}

// consumer

@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() {
    MqttPahoMessageDrivenChannelAdapter adapter = new MqttPahoMessageDrivenChannelAdapter("siSampleConsumer",
            mqttClientFactory(), "siSampleTopic");
    adapter.setCompletionTimeout(5000);
    adapter.setConverter(new DefaultPahoMessageConverter());
    adapter.setQos(1);
    return adapter;
}

如果您需要在发布消息后等待答案/结果，则使用 2 个单独的连接会变得很困难...

Answer 1

TL;DR

答案是否定的，当前的 Spring Boot MQTT 集成实现不会（甚至未来的实现也可能不会）。

回答

我面临着同样的情况：我需要一个 MQTT 客户端在入站和出站中都打开，使连接持久并共享相同的配置（客户端 ID、凭据等），使用 Spring Integration Flows 作为尽可能接近设计。

为了实现这一点，我不得不重新实现 MqttPahoMessageDrivenChannelAdapter 和 MqttPahoMessageHandler 以及一个客户端工厂。

在MqttPahoMessageDrivenChannelAdapter 和MqttPahoMessageHandler 中，我必须选择使用异步的（IMqttAsyncClient）来修复使用哪一个。然后我必须查看调用/使用客户端实例的部分代码，以检查它是否已被其他流实例化并检查状态（例如，如果它已经连接，则不要尝试连接它）。

关于客户端工厂，它更容易：我重新实现了getAsyncClientInstance(String url, String clientId)，使用url 和clientId 的串联作为哈希键，将实例存储到用于检索现有实例的映射中如果其他流程请求它。

它以某种方式起作用，但这只是一个测试，我什至不确定这是一个好方法。 (I've started another StackOverflow question in order to track my specific scenario)。

你能分享一下你是如何处理你的情况的吗？

Answer 2

第一个连接会断开，消息发送后会重新连接。

不知道你的意思；两个组件都将保持打开一个持久连接。

由于工厂不连接客户端，适配器连接，它不是为使用共享客户端而设计的。

使用单个连接并不能真正帮助协调请求/回复，因为回复仍会在另一个线程上异步返回。

如果您在请求/回复中有一些数据可用于关联对请求的回复，您可以使用BarrierMessageHandler 来执行该任务。示例见my answer here；它使用标准的相关 id 标头，但 MQTT 无法做到这一点，您需要在消息中添加一些内容。