Spring boot + MQTT

系统管理员 2022-05-16 05:44 111阅读 0赞

Springboot 整合MQTT。

MQTT服务器采用的是Apache ActiveMQ。

启动Apache ActiveMQ会有多个传输协议，这里采用的是tcp的方式，默认端口1883

1.创建配置文件config.properties

mqtt.host=tcp://127.0.0。1:1883
    mqtt.clientid=mqttjs_e8022a4d0b
    mqtt.username=admin
    mqtt.password=admin
    mqtt.topic=topic1
    mqtt.timeout=1000
    mqtt.keepalive=2000

2.创建MQTT相关配置类

public class MqttPushClient {
        private static final Logger LOGGER = LoggerFactory.getLogger(MqttPushClient.class);
        private static MqttClient client;
    
        public static MqttClient getClient() {
            return client;
        }
    
        public static void setClient(MqttClient client) {
            MqttPushClient.client = client;
        }
    
        public void connect(String host, String clientID, String username, String password, int timeout, int keepalive){
            MqttClient client;
            try {
                client = new MqttClient(host, clientID, new MemoryPersistence());
                MqttConnectOptions options = new MqttConnectOptions();
                options.setCleanSession(false);
                options.setUserName(username);
                options.setPassword(password.toCharArray());
                options.setConnectionTimeout(timeout);
                options.setKeepAliveInterval(keepalive);
                MqttPushClient.setClient(client);
                try {
                    client.setCallback(new PushCallback());
                    client.connect(options);
                } catch (Exception e) {
                    e.printStackTrace();
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    
        /**
         * 发布，默认qos为0，非持久化
         * @param topic
         * @param pushMessage
         */
        public void publish(String topic,String pushMessage){
            publish(0, false, topic, pushMessage);
        }
    
        /**
         * 发布
         * @param qos
         * @param retained
         * @param topic
         * @param pushMessage
         */
        public void publish(int qos,boolean retained,String topic,String pushMessage){
            MqttMessage message = new MqttMessage();
            message.setQos(qos);
            message.setRetained(retained);
            message.setPayload(pushMessage.getBytes());
            MqttTopic mTopic = MqttPushClient.getClient().getTopic(topic);
            if(null == mTopic){
                LOGGER.error("topic not exist");
            }
            MqttDeliveryToken token;
            try {
                token = mTopic.publish(message);
                token.waitForCompletion();
            } catch (MqttPersistenceException e) {
                e.printStackTrace();
            } catch (MqttException e) {
                e.printStackTrace();
            }
        }
    
        /**
         * 订阅某个主题，qos默认为0
         * @param topic
         */
        public void subscribe(String topic){
            subscribe(topic,0);
        }
    
        /**
         * 订阅某个主题
         * @param topic
         * @param qos
         */
        public void subscribe(String topic,int qos){
            try {
                MqttPushClient.getClient().subscribe(topic, qos);
            } catch (MqttException e) {
                e.printStackTrace();
            }
        }
    }

3.Spring boot 读取自定义properties配置文件，同时初始化配置类

@Component
    @Configuration
    @PropertySource("classpath:config.properties")
    @ConfigurationProperties(prefix = "mqtt")
    public class MqttConfiguration {
        private String host;
        private String clientid;
        private String username;
        private String password;
        private String topic;
        private int timeout;
        private int keepalive;
    
        public String getClientid() {
            return clientid;
        }
    
        public void setClientid(String clientid) {
            this.clientid = clientid;
        }
    
        public String getUsername() {
            return username;
        }
    
        public void setUsername(String username) {
            this.username = username;
        }
    
        public String getPassword() {
            return password;
        }
    
        public void setPassword(String password) {
            this.password = password;
        }
    
        public String getTopic() {
            return topic;
        }
    
        public void setTopic(String topic) {
            this.topic = topic;
        }
    
        public int getTimeout() {
            return timeout;
        }
    
        public void setTimeout(int timeout) {
            this.timeout = timeout;
        }
    
        public int getKeepalive() {
            return keepalive;
        }
    
        public void setKeepalive(int keepalive) {
            this.keepalive = keepalive;
        }
    
        public String getHost() {
            return host;
        }
    
        public void setHost(String host) {
            this.host = host;
        }
        @Bean
        public MqttPushClient getMqttPushClient(){
            MqttPushClient mqttPushClient = new MqttPushClient();
            mqttPushClient.connect(host, clientid, username, password, timeout,keepalive);
            return mqttPushClient;
        }
    }

4.消费监听类

public class PushCallback implements MqttCallback {
    
        @Override
        public void connectionLost(Throwable cause) {
            // 连接丢失后，一般在这里面进行重连
            System.out.println("连接断开，可以做重连");
        }
    
        @Override
        public void deliveryComplete(IMqttDeliveryToken token) {
            System.out.println("deliveryComplete---------" + token.isComplete());
        }
    
        @Override
        public void messageArrived(String topic, MqttMessage message) throws Exception {
            // subscribe后得到的消息会执行到这里面
            System.out.println("接收消息主题 : " + topic);
            System.out.println("接收消息Qos : " + message.getQos());
            System.out.println("接收消息内容 : " + new String(message.getPayload()));
        }
    }

5.controller样例

@Controller
    public class MqttController {
        @Autowired
        private MqttPushClient mqttPushClient;
        @RequestMapping("/hello")
        @ResponseBody
        public String sendHello(){
            String kdTopic = "topic1";
            mqttPushClient.publish(0, false, kdTopic, "15345715326");
            mqttPushClient.subscribe(kdTopic);
            return "123";
        }
    }