Spring Integration 接入 mqtt 协议

可以使用 Spring Integration 的 MQTT 模块来实现 MQTT 协议的接入。

首先，需要在项目中添加 Spring Integration MQTT 模块的依赖：

<dependency>
    <groupId>org.springframework.integration</groupId>
    <artifactId>spring-integration-mqtt</artifactId>
    <version>5.5.0</version>
</dependency>

然后，可以使用 Spring Integration 提供的 MqttPahoMessageDrivenChannelAdapter 来实现 MQTT 消息的接收。示例代码如下：

@Configuration
@EnableIntegration
public class MqttConfiguration {

    @Bean
    public MqttPahoClientFactory mqttClientFactory() {
        DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
        MqttConnectOptions options = new MqttConnectOptions();
        // 配置 MQTT 连接参数
        factory.setConnectionOptions(options);
        return factory;
    }

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

    @Bean
    public MessageProducer inbound() {
        MqttPahoMessageDrivenChannelAdapter adapter =
                new MqttPahoMessageDrivenChannelAdapter("clientId", mqttClientFactory(), "topic");
        adapter.setCompletionTimeout(5000);
        adapter.setConverter(new DefaultPahoMessageConverter());
        adapter.setQos(1);
        adapter.setOutputChannel(mqttInputChannel());
        return adapter;
    }

    @ServiceActivator(inputChannel = "mqttInputChannel")
    public void handleMessage(Message<String> message) {
        // 处理接收到的 MQTT 消息
        System.out.println("Received MQTT Message: " + message.getPayload());
    }
}

上述代码中，配置了一个 MqttPahoMessageDrivenChannelAdapter，用于监听指定的 MQTT topic，并将接收到的消息转换为 Spring Integration 的 Message 对象，并发送到指定的消息通道 mqttInputChannel 中。同时，还定义了一个 ServiceActivator，用于处理接收到的消息。

需要注意的是，配置 MqttPahoMessageDrivenChannelAdapter 时，需要指定 clientId、mqttClientFactory 和 topic 参数，其中 clientId 为客户端标识符，mqttClientFactory 为 MQTT 客户端工厂，可以通过配置 MqttConnectOptions 来设置连接参数，topic 为要监听的 MQTT topic。

除了接收 MQTT 消息外，Spring Integration 还提供了 MqttPahoMessageHandler 类，可用于发送 MQTT 消息。使用方法与 MqttPahoMessageDrivenChannelAdapter 类似，只需将其配置为 MessageProducer，并将要发送的消息发送到指定的消息通道即可。