Springboot整合Mqtt

Spring Boot整合MQTT需要使用Spring Integration MQTT支持。以下是一个基本的例子：

1. 添加依赖到你的pom.xml：

<dependencies>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-integration</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.integration</groupId>
        <artifactId>spring-integration-mqtt</artifactId>
    </dependency>
</dependencies>

2. 配置MQTT客户端并定义消息通道：

@Configuration
public class MqttConfig {
 
    @Value("${mqtt.broker.url}")
    private String brokerUrl;
 
    @Value("${mqtt.client.id}")
    private String clientId;
 
    @Value("${mqtt.username}")
    private String userName;
 
    @Value("${mqtt.password}")
    private String password;
 
    @Value("${mqtt.default.topic}")
    private String defaultTopic;
 
    @Bean
    public MqttPahoClientFactory mqttClient() {
        DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
        MqttConnectOptions options = new MqttConnectOptions();
        options.setServerURIs(new String[]{brokerUrl});
        options.setUserName(userName);
        options.setPassword(password.toCharArray());
        factory.setConnectionOptions(options);
        return factory;
    }
 
    @Bean
    public MessageChannel mqttInputChannel() {
        return new DirectChannel();
    }
 
    @Bean
    public MessageProducer inbound() {
        MqttPahoMessageDrivenChannelAdapter adapter = 
          new MqttPahoMessageDrivenChannelAdapter(clientId, mqttClient(), defaultTopic);
        adapter.setCompletionTimeout(5000);
        adapter.setQos(2);
        adapter.setOutputChannel(mqttInputChannel());
        return adapter;
    }
}

3. 监听MQTT通道并处理消息：

@Component
public class MqttReceiver {
 
    @Autowired
    private MessageChannel mqttInputChannel;
 
    @Autowired
    private SimpMessagingTemplate simpMessagingTemplate;
 
    @PostConstruct
    public void init() {
        mqttInputChannel.subscribe(Message.class);
    }
 
    @MessageMapping("mqtt")
    public void receiveMessage(String payload) {
        // 处理接收到的消息
        System.out.println("Received Message: " + payload);
        // 可以发送消息到其他通道或者应用
        simpMessagingTemplate.convertAndSend("/topic/mqtt", payload);
    }
}

确保你的application.properties或application.yml文件包含了正确的MQTT配置：

mqtt.broker.url=tcp://localhost:1883
mqtt.client.id=testClient
mqtt.usernam