MQTT工具类

逃离我推掉我的手 2024-03-17 17:51 8阅读 0赞

### 1、MQTT工具类 ###

import lombok.extern.slf4j.Slf4j;
    import org.apache.dubbo.common.constants.QosConstants;
    import org.eclipse.paho.client.mqttv3.MqttCallback;
    import org.eclipse.paho.client.mqttv3.MqttClient;
    import org.eclipse.paho.client.mqttv3.MqttException;
    import org.eclipse.paho.client.mqttv3.MqttMessage;
    
    /**
     * mqtt 发布订阅工具类
     */
    @Slf4j
    public class MqttClientUtil {
        
    
        private MqttClient mqttClient;
    
        public MqttClientUtil(MqttClient mqttClient) {
        
            this.mqttClient = mqttClient;
        }
    
        /**
         * 订阅mqtt消息
         *
         * @param topic        主题
         * @param mqttCallback 回调函数
         */
        public void subscribe(String topic, MqttCallback mqttCallback) {
        
            try {
        
                mqttClient.setCallback(mqttCallback);
                mqttClient.subscribe(topic);
            } catch (MqttException e) {
        
                log.error("mqtt订阅消息失败！topic：{}", topic);
                e.printStackTrace();
            }
        }
    
        /**
         * 常用发布mqtt消息
         *
         * @param topic   主题
         * @param message 消息内容
         */
        public void publishSimple(String topic, String message) {
        
            publish(topic, message, 1, true);
        }
    
        /**
         * 发布mqtt消息
         *
         * @param topic    主题
         * @param message  消息内容
         * @param qos      QoS级别 0：最多一次, 1：至少一次, 2：仅一次
         * @param retained 是否保留最新消息 true是，false否
         */
        public void publish(String topic, String message, int qos, boolean retained) {
        
        	try {
        
                MqttMessage mqttMessage = new MqttMessage(message.getBytes());
                mqttMessage.setQos(qos);
                mqttMessage.setRetained(retained);
                mqttClient.publish(topic, mqttMessage);
            } catch (MqttException e) {
        
                if ("客户机未连接".equals(e.getMessage())){
        
                    try {
        
                        mqttClient.connect();
                    }catch (Exception exception){
        
                        log.error("客户机重连失败！");
                    }
                    // 重连后重新发送
                    publish(topic, message, qos, retained);
                    return;
                }
                log.error("mqtt发布消息失败！topic：{},message：{}", topic, message);
                e.printStackTrace();
            }
        }
    }

### 2、MQTT配置类 ###

import org.eclipse.paho.client.mqttv3.MqttClient;
    import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
    import org.eclipse.paho.client.mqttv3.MqttException;
    import org.springframework.beans.factory.annotation.Value;
    import org.springframework.context.annotation.Bean;
    import org.springframework.stereotype.Component;
    
    /**
     * @Filename: MqttConfig mqtt配置类
     * @Author: sheng.wanping
     * <li>Date: 2023/7/6 17:13
     * <li>Version: 1.0
     * <li>Content: create
     */
    @Component
    public class MqttConfig {
        
    
        @Value("${spring.mqtt.url}")
        private String brokerUrl; // MQTT服务器地址
    
        @Value("${spring.mqtt.username}")
        private String username; // MQTT服务器用户名
    
        @Value("${spring.mqtt.password}")
        private String password; // MQTT服务器密码
    
        private final String CLIENT_ID = "mqttHook_client_id"; // 客户端ID
    
        /**
         * @return 注册一个 MqttClientUtil 的bean
         * @throws MqttException
         */
        @Bean
        public MqttClientUtil getMqttClient() throws MqttException {
        
            MqttClient client = new MqttClient(brokerUrl, CLIENT_ID);
            MqttConnectOptions options = new MqttConnectOptions();
            options.setUserName(username);
            options.setPassword(password.toCharArray());
            client.connect(options);
            return new MqttClientUtil(client);
        }
    }

### 3、回调函数 Callback ###

import lombok.extern.slf4j.Slf4j;
    import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
    import org.eclipse.paho.client.mqttv3.MqttCallback;
    import org.eclipse.paho.client.mqttv3.MqttMessage;
    import org.springframework.stereotype.Component;
    
    @Slf4j
    @Component
    public class MyMqttCallback implements MqttCallback {
        
    
        @Override
        public void connectionLost(Throwable cause) {
        
            // 处理连接丢失事件,可以添加重连处理
            log.error("mqtt连接丢失！");
        }
    
        @Override
        public void messageArrived(String topic, MqttMessage message) throws Exception {
        
            // 处理接收到的消息
            String payload = new String(message.getPayload());
            System.out.println("接收到消息：" + payload);
            // 处理业务逻辑
        }
    
        @Override
        public void deliveryComplete(IMqttDeliveryToken token) {
        
            // 发布消息到mqtt服务器后，会调用该方法。
            // 处理消息发送完成事件
            System.out.println("发布消息成功");
        }
    }

### 4、测试 ###

@RestController
    @Api(tags = "测试mqtt")
    @RequestMapping("/test")
    public class TestController {
        
    
      @Autowired private MqttClientUtil mqttClientUtil;
    
      @PostMapping("/test")
      @ApiOperation("测试")
      public void test() {
        
        String topic = "testtopic/ha";
        mqttClientUtil.subscribe(topic, new MyMqttCallback());
        System.out.println("成功");
    
        mqttClientUtil.publishSimple(topic, "哈哈哈");
        System.out.println("成功2");
      }
    }