SpringBoot 使用 MQTT 发送消息

     		<!-- mqtt -->
        <dependency>
            <groupId>org.springframework.integration</groupId>
            <artifactId>spring-integration-stream</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.integration</groupId>
            <artifactId>spring-integration-mqtt</artifactId>
        </dependency>
        <!-- mqtt end -->

    #mqtt
    mqtt:
      host: tcp://***.***.**.**:1883
      clientinid: test
      topic: topic
      qoslevel: 2
      username:
      password:
      async: true
      keepaliveinterval: 2

    import org.apache.commons.lang3.StringUtils;
    import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
    import org.springframework.beans.factory.annotation.Value;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    import org.springframework.integration.annotation.IntegrationComponentScan;
    import org.springframework.integration.annotation.ServiceActivator;
    import org.springframework.integration.channel.DirectChannel;
    import org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory;
    import org.springframework.integration.mqtt.core.MqttPahoClientFactory;
    import org.springframework.integration.mqtt.outbound.MqttPahoMessageHandler;
    import org.springframework.messaging.MessageChannel;
    import org.springframework.messaging.MessageHandler;
    
    @Configuration
    @IntegrationComponentScan
    public class MqttConfig {
    
    @Value("${mqtt.username}")
    private String username;
    
    @Value("${mqtt.password}")
    private String password;
    
    @Value("${mqtt.host}")
    private String hostUrl;
    
    @Value("${mqtt.clientinid}")
    private String clientId;
    
    @Value("${mqtt.topic}")
    private String defaultTopic;
    
    @Value("${mqtt.qoslevel}")
    private int qosLevel;
    
    @Value("${mqtt.async}")
    private boolean async;
    
    @Value("${mqtt.keepaliveinterval}")
    private int keepAliveInterval;
    
    public MqttConnectOptions getMqttConnectOptions(){
        MqttConnectOptions mqttConnectOptions = new MqttConnectOptions();
        if (StringUtils.isNotBlank(username)) {
            mqttConnectOptions.setUserName(username);
        }
        if (StringUtils.isNotBlank(password)) {
            mqttConnectOptions.setPassword(password.toCharArray());
        }
        mqttConnectOptions.setServerURIs(new String[]{hostUrl});
        mqttConnectOptions.setKeepAliveInterval(keepAliveInterval);
        return mqttConnectOptions;
    }
    
    @Bean
    public MqttPahoClientFactory mqttClientFactory() {
        DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
        factory.setConnectionOptions(getMqttConnectOptions());
        return factory;
    }
    
    // 初始化
    @Bean
    @ServiceActivator(inputChannel = "mqttOutboundChannel")
    public MessageHandler mqttOutbound() {
        MqttPahoMessageHandler messageHandler = new MqttPahoMessageHandler(clientId, mqttClientFactory());
        messageHandler.setAsync(async);
        messageHandler.setDefaultTopic(defaultTopic);
        messageHandler.setDefaultQos(qosLevel);
        return messageHandler;
    }
    
    @Bean
    public MessageChannel mqttOutboundChannel() {
        return new DirectChannel();
    }
    }

    import org.springframework.integration.annotation.MessagingGateway;
    import org.springframework.integration.mqtt.support.MqttHeaders;
    import org.springframework.messaging.handler.annotation.Header;
    
    @MessagingGateway(defaultRequestChannel = "mqttOutboundChannel")
    public interface MqttGateway {
    void sendToMqtt(String data,@Header(MqttHeaders.TOPIC) String topic);
    }