SpringBoot整合MQTT

绝地灬酷狼 2023-01-19 09:51 15阅读 0赞

### SpringBoot整合MQTT ###

#### 1、导包 ####

<dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-starter-integration</artifactId>
                <version> 2.1.12.RELEASE</version>
            </dependency>
            <dependency>
                <groupId>org.springframework.integration</groupId>
                <artifactId>spring-integration-stream</artifactId>
                <version> 5.1.9.RELEASE</version>
            </dependency>
            <dependency>
                <groupId>org.springframework.integration</groupId>
                <artifactId>spring-integration-mqtt</artifactId>
                <version> 5.1.9.RELEASE</version>
            </dependency>

配置信息

server:
      port: 9000
    
    spring:
      application:
        name: mqtt-center
      mqtt:
        username: admin
        password: admin
        url: tcp://127.0.0.1:1883
        client:
          id: mqttId
        default:
          topic: foo
        completionTimeout: 3000

#### 2、编写客户端 ####

@Configuration
    public class MqttConfig { 
    
        private static final byte[] WILL_DATA;
    
        static { 
            WILL_DATA = "offline".getBytes();
        }
    
        @Value("${spring.mqtt.username}")
        private String username;
    
        @Value("${spring.mqtt.password}")
        private String password;
    
        @Value("${spring.mqtt.url}")
        private String hostUrl;
    
        @Value("${spring.mqtt.client.id}")
        private String clientId;
    
        @Value("${spring.mqtt.default.topic}")
        private String defaultTopic;
    
        @Value("${spring.mqtt.completionTimeout}")
        private int completionTimeout;   
    
        /** * MQTT连接器选项配置 * **/
        @Bean(value = "getMqttConnectOptions")
        public MqttConnectOptions getMqttConnectOptions(){ 
            MqttConnectOptions mqttConnectOptions=new MqttConnectOptions();
            // 设置是否清空session,这里如果设置为false表示服务器会保留客户端的连接记录，这里设置为true表示每次连接到服务器都以新的身份连接
            mqttConnectOptions.setCleanSession(true);
            // 设置超时时间 单位为秒
            mqttConnectOptions.setConnectionTimeout(10);
            mqttConnectOptions.setAutomaticReconnect(true);
            mqttConnectOptions.setUserName(username);
            mqttConnectOptions.setPassword(password.toCharArray());
            mqttConnectOptions.setServerURIs(new String[]{ hostUrl});
            // 设置会话心跳时间 单位为秒 服务器会每隔1.5*20秒的时间向客户端发送心跳判断客户端是否在线，但这个方法并没有重连的机制
            mqttConnectOptions.setKeepAliveInterval(10);
    
            return mqttConnectOptions;
        }
    
        /** * MQTT客户端工程类 * **/
        @Bean
        public MqttPahoClientFactory mqttClientFactory() { 
            DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
            factory.setConnectionOptions(getMqttConnectOptions());
            return factory;
        }
    
    
        /** * MQTT发送消息通道 * **/
        @Bean
        public MessageChannel mqttOutboundChannel() { 
            return new DirectChannel();
        }
    
    
        /** * 配置MQTT消息推送器 * **/
        @Bean
        @ServiceActivator(inputChannel = "mqttOutboundChannel")
        public MessageHandler mqttOutbound() { 
            MqttPahoMessageHandler messageHandler =  new MqttPahoMessageHandler(clientId, mqttClientFactory());
            messageHandler.setAsync(true);
            messageHandler.setDefaultTopic(defaultTopic);
            return messageHandler;
        }
    
    
        /** * MQTT接受信息通道 * **/
        @Bean
        public MessageChannel mqttInputChannel() { 
            return new DirectChannel();
        }
    
        /** * 配置client,监听的topic * MQTT消息订阅绑定（消费者） * 配置消息适配器，配置订阅客户端 **/
        @Bean
        public MessageProducer inbound() { 
            MqttPahoMessageDrivenChannelAdapter adapter =
                    new MqttPahoMessageDrivenChannelAdapter(clientId + "_inbound1", mqttClientFactory(),
                            defaultTopic);
            adapter.setCompletionTimeout(completionTimeout);
            adapter.setConverter(new DefaultPahoMessageConverter());
            adapter.setQos(2);
            adapter.setOutputChannel(mqttInputChannel());
            return adapter;
        }
    
        /** * MQTT消息处理器 * **/
        @Bean
        @ServiceActivator(inputChannel = "mqttInputChannel")
        public MessageHandler handler() { 
            return new MessageHandler() { 
                public void handleMessage(Message message) throws MessagingException {  
                    //处理接收消息
                    System.out.println(message);
    
                }
            };
        }
    
    }

#### 3、编写网关 ####

@MessagingGateway(defaultRequestChannel = "mqttOutboundChannel")
    public interface MqttGateway { 
     
        /** * 发送信息到MQTT服务器 * * @param data 发送的文本 */
        void sendToMqtt(String data);
     
        /** * 发送信息到MQTT服务器 * * @param topic 主题 * @param payload 消息主体 */
        void sendToMqtt(@Header(MqttHeaders.TOPIC) String topic,
                        String payload);
     
        /** * 发送信息到MQTT服务器 * * @param topic 主题 * @param qos 对消息处理的几种机制。 * 0 表示的是订阅者没收到消息不会再次发送，消息会丢失。 * 1 表示的是会尝试重试，一直到接收到消息，但这种情况可能导致订阅者收到多次重复消息。 * 2 多了一次去重的动作，确保订阅者收到的消息有一次。 * @param payload 消息主体 */
        void sendToMqtt(@Header(MqttHeaders.TOPIC) String topic,
                        @Header(MqttHeaders.QOS) int qos,
                        String payload);
    }

#### 4、调用 ####

@RestController
    public class MqttController { 
        @Autowired
        private MqttGateway mqttGateway;
    
        @GetMapping("test")
        public String test(){ 
            mqttGateway.sendToMqtt("foo","发送消息");
            return "消息发送成功";
        }
    }

经过上面配置，我们就可以搭配MQTT.fx进行简单的消息发送。