Springboot 写一个mqtt 发布/订阅案例(二)

骑猪看日落 2023-02-15 03:42 20阅读 0赞

其它的相关配置参考我这篇文章：[Springboot 写一个mqtt 发布/订阅案例(一)][Springboot _mqtt]，在这篇文章中，就贴代码了。

##### 实现的功能是：将“订阅的数据” 同时“发布”出去。 #####

### 1.1 添加依赖 ###

<dependency>
        <groupId>org.eclipse.paho</groupId>
        <artifactId>org.eclipse.paho.client.mqttv3</artifactId>
        <version>1.2.0</version>
    </dependency>
    
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-integration</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.integration</groupId>
        <artifactId>spring-integration-stream</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.integration</groupId>
        <artifactId>spring-integration-mqtt</artifactId>
    </dependency>
    <dependency>
        <groupId>com.google.code.gson</groupId>
        <artifactId>gson</artifactId>
        <version>2.8.2</version>
    </dependency>

### 1.2 application.properties配置类 ###

mqtt.broker = tcp://177.177.7.177:1777
    mqtt.clientId = ups_mqtt_service
    mqtt.userName = client
    mqtt.password = client
    mqtt.connectionTimeout = 3
    mqtt.keepAliveInterval = 10
    
    spring.jackson.serialization.FAIL_ON_EMPTY_BEANS = false
    eureka.client.serviceUrl.defaultZone = tcp://177.177.7.177:1777
    management.security.enabled = false
    logging.file.name = /ROOT/logs/ups.log
    
    access.server.run-action-in-queue = true
    access.server.port = 8027
    hystrix.command.default.execution.isolation.thread.timeoutInMilliseconds = 15000
    
    apollo.bootstrap.enabled = true
    eureka.instance.perfer-ip-address = true
    spring.application.name=mqtt_ups_service_test

### 1.3 启动类 ###

package com.mqtt_jowoiot.mqtt_jowoiot;
    
    import org.eclipse.paho.client.mqttv3.MqttException;
    import org.slf4j.LoggerFactory;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.boot.SpringApplication;
    import org.springframework.boot.autoconfigure.SpringBootApplication;
    import javax.annotation.PostConstruct;
    
    @SpringBootApplication
    public class MqttJowoiotApplication {
        
    
        @Autowired
        private JowoiotMqttClient jowoiotMqttClient;
    
        //@Override
        @PostConstruct
        public void run() throws MqttException {
        
            LoggerFactory.getLogger(MqttJowoiotApplication.class).info("ups server start...");
    //        ScheduledExecutorService executorService = new ScheduledThreadPoolExecutor(1);
            jowoiotMqttClient.connect();
            jowoiotMqttClient.subscrib();
            jowoiotMqttClient.publish();
        }
    
    
        public static void main(String[] args) {
        
            SpringApplication.run(MqttJowoiotApplication.class, args);
        }
    
    }

### 1.4 新建BeanConfig类 ###

package com.mqtt_jowoiot.mqtt_jowoiot;
    
    import org.eclipse.paho.client.mqttv3.MqttClient;
    import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
    import org.eclipse.paho.client.mqttv3.MqttException;
    import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
    import org.springframework.beans.factory.annotation.Value;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    
    import java.util.UUID;
    
    /**
     * mqtt客户端连接配置信息
     */
    @Configuration
    public class BeanConfig {
        
    
        @Value("${mqtt.broker:tcp://177.177.7.177:1777}")
        String broker;
    
        /*@Value("${mqtt.clientId:ups_mqtt_service}")
        String clientId;*/
    
        @Value("${mqtt.userName:client}")
        String username;
    
        @Value("${mqtt.password:client}")
        String password;
    
        @Value("${mqtt.connectionTimeout:3}")
        int connectTimeout;
    
        @Value("${mqtt.keepAliveInterval:10}")
        int keepAliveInterval;
    
    
        @Bean
        public MqttConnectOptions mqttConnectOptions() {
        
            MqttConnectOptions options = new MqttConnectOptions();
            options.setCleanSession(false);
            options.setAutomaticReconnect(true);
            options.setConnectionTimeout(connectTimeout);
            options.setKeepAliveInterval(keepAliveInterval);
            options.setPassword(password.toCharArray());
            options.setUserName(username);
            options.setServerURIs(new String[]{
        broker});
            return options;
        }
    
        @Bean
        public MqttClient mqttClient() throws MqttException {
        
            return new MqttClient(broker,"UPS_"+ UUID.randomUUID().toString(),new MemoryPersistence());
    
        }
    }

### 1.5 新建JowoiotMqttClient类 ###

package com.mqtt_jowoiot.mqtt_jowoiot;
    
    
    import org.eclipse.paho.client.mqttv3.*;
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.beans.factory.annotation.Value;
    import org.springframework.stereotype.Component;
    
    import java.util.ArrayList;
    import java.util.HashMap;
    import java.util.List;
    import java.util.Map;
    import java.util.concurrent.ConcurrentLinkedQueue;
    import java.util.concurrent.TimeUnit;
    
    
    @Component
    public class JowoiotMqttClient {
        
    
        Logger logger = LoggerFactory.getLogger(JowoiotMqttClient.class.getName());
    
        //public static Set<String> topicHeartbeats = new HashSet<>();
        public static String topicToServer;
    
        @Autowired
        MqttClient mqttClient;
    
        @Autowired
        MqttConnectOptions mqttConnectOptions;
    
        public static ConcurrentLinkedQueue<List<Map<String, Object>>> payloadQueue = new ConcurrentLinkedQueue<>();
    
    
        @Value("${mqtt.qos:2}")
        int qos;
    
        @Value("${retained:false}")
        boolean retained;
    
        public void connect() throws MqttException {
        
            mqttClient.connect(mqttConnectOptions);
        }
    
        //订阅消息
        public void subscrib() {
        
            try {
        
                mqttClient.subscribe("jowoiot/toServer/bruce/lptestsub");
                mqttClient.setCallback(new MqttCallback() {
        
                    @Override
                    public void connectionLost(Throwable throwable) {
        
                        System.out.println("connectionLost");
                    }
    
                    @Override
                    public void messageArrived(String s, MqttMessage mqttMessage) throws Exception {
        
                        System.out.println("s:" + s);
                        //接收消息
                        byte[] payload = mqttMessage.getPayload();
                        String message = new String(payload, "utf-8");
                        System.out.println("message:" + message);
                        //处理消息
                        List<Map<String, Object>> mapList = new ArrayList<>();
                        Map<String, Object> map = new HashMap<>();
                        map.put("message", message);
                        mapList.add(map);
                        //数据放入容器payloadQueue
                        payloadQueue.offer(mapList);
                    }
    
                    @Override
                    public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {
        
                        System.out.println("deliveryComplete");
                    }
                });
            } catch (MqttException e) {
        
    
                e.printStackTrace();
            }
    
    
        }
    
        /**
         * 发布消息
         * @throws MqttException
         */
        public void publish() throws MqttException {
        
            while (true) {
        
                try {
        
                    TimeUnit.SECONDS.sleep(2);
                    //取出messageMap数据
                    List<Map<String, Object>> poll = payloadQueue.poll();
                    //发送数据
                    System.out.println("poll.toString().getBytes(): " );// 打印需要发送的数据
                    if (poll != null) {
        
                        mqttClient.publish("jowoiot/toServer/bruce/lptestpub", poll.toString().getBytes(), qos, retained);
                        logger.info("publish success :\t topicToServer:" + "jowoiot/toServer/bruce/lptestpub" + "\npayload:" + poll.toString());
                    }
                } catch (InterruptedException e) {
        
                    e.printStackTrace();
                }
            }
    
        }
    }

[Springboot _mqtt]: https://blog.csdn.net/qq_37788383/article/details/106500075