如何使用Spring integration在Springboot中集成Mqtt详解

如何使用Spring Integration在Spring Boot中集成MQTT？

Spring Integration的Mqtt模块提供了使用Java Mqtt客户端连接到MQTT代理的模板类、消息驱动通道适配器，在Spring Boot应用程序中非常容易集成。下面是使用Spring Integration在Spring Boot中集成MQTT的完整攻略。

第一步：添加依赖

Spring Boot提供了starter-pom来快速创建应用程序并引入所需的依赖。如果需要使用Spring Integration Mqtt模块，需要在pom.xml中添加以下依赖。

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

第二步：配置Mqtt连接

在application.properties文件中配置Mqtt连接。

spring.mqtt.username=username
spring.mqtt.password=password
spring.mqtt.url=tcp://localhost:1883

第三步：定义MqttInboundChannelAdapter

在Spring Integration中，MqttInboundChannelAdapter用于从Mqtt服务器接收消息，并将它们投递到通道中。

@Configuration
@EnableIntegration
@IntegrationComponentScan
public class MqttConfiguration {

    @Bean
    public MqttPahoClientFactory mqttClientFactory() {
        DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
        factory.setUserName(username);
        factory.setPassword(password);
        factory.setServerURIs(url);
        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;
    }

    @Bean
    public MessageHandler echoHandler() {
        return new MessageHandler() {

            @Override
            public void handleMessage(Message<?> message) throws MessagingException {
                System.out.println(message.getPayload());
            }

        };
    }

    @Bean
    public IntegrationFlow mqttInputFlow() {
        return IntegrationFlows.from(mqttInputChannel())
                .handle(echoHandler())
                .get();
    }
}

第四步：发送Mqtt消息

发送Mqtt消息很容易。可以使用MqttPahoMessageHandler类将消息注入到Mqtt通道中。

@Autowired
private IntegrationFlow mqttOutFlow;

@Autowired
private MessageChannel mqttOutboundChannel;


private void sendMessage(String payload) {
    Message<String> message = MessageBuilder.withPayload(payload).build();
    this.mqttOutboundChannel().send(message);
}

下面是一个完整示例。

@RestController
public class MqttExample {

    @Autowired
    private MessageChannel mqttOutboundChannel;

    @RequestMapping(value = "/send", method = RequestMethod.GET)
    public String sendMessage() {
        this.sendMessage("Hello, MQTT!");
        return "Message sent to MQTT";
    }

    private void sendMessage(String payload) {
        Message<String> message = MessageBuilder.withPayload(payload).build();
        this.mqttOutboundChannel().send(message);
    }

}

在以上代码中，如果访问/endpoint，则会向MQTT服务器发送一条消息，消息内容为"Hello, MQTT!"。

第五步：运行应用程序

现在可以启动应用程序并向MQTT代理发送和接收消息了。现在，如果您想向MQTT代理发送消息，只需访问http://localhost:8080/send即可。

如果MQTT代理成功收到消息，您将在控制台中看到打印的消息。

这就是一个完整的使用Spring Integration在Spring Boot中集成MQTT的攻略。