mqtt工具类

@@ -84,6 +84,11 @@
             <artifactId>jedis</artifactId>
             <artifactId>jedis</artifactId>
             <version>2.9.0</version>
             <version>2.9.0</version>
         </dependency>
         </dependency>
+        <dependency>
+            <groupId>org.eclipse.paho</groupId>
+            <artifactId>org.eclipse.paho.client.mqttv3</artifactId>
+            <version>1.2.5</version>
+        </dependency>
     </dependencies>
     </dependencies>
 
 
     <build>
     <build>

+package com.iot.scheduler.utils;
+
+import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.StringUtils;
+import org.eclipse.paho.client.mqttv3.*;
+import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
+
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.UUID;
+
+/**
+ * MQTT客户端工具类
+ * 支持连接、发布、订阅、断开连接等操作
+ */
+@Slf4j
+public class MqttUtils {
+
+    private MqttClient mqttClient;
+    private MqttConnectOptions connectOptions;
+    private ConnectionCallback connectionCallback;
+    private MessageCallback messageCallback;
+    private boolean isConnected = false;
+    private List<String> subscribedTopics = new ArrayList<>();
+
+    /**
+     * 连接回调接口
+     */
+    public interface ConnectionCallback {
+        void onConnected();
+
+        void onDisconnected();
+
+        void onConnectionFailed(Throwable cause);
+
+        void onConnectionLost(Throwable cause);
+    }
+
+    /**
+     * 消息回调接口
+     */
+    public interface MessageCallback {
+        void onMessageReceived(String topic, String message);
+
+        void onMessageDelivered(IMqttDeliveryToken token);
+    }
+
+    /**
+     * MQTT连接配置构建器
+     */
+    public static class MqttConfig {
+        private String broker;
+        private String clientId;
+        private String username;
+        private String password;
+        private int connectionTimeout = 10;
+        private int keepAliveInterval = 60;
+        private boolean cleanSession = true;
+        private boolean automaticReconnect = true;
+        private int maxInflight = 10;
+        private int qos = 1;
+
+        public MqttConfig(String broker) {
+            this.broker = broker;
+            this.clientId = "Client_" + UUID.randomUUID().toString().substring(0, 8);
+        }
+
+        public MqttConfig setBroker(String broker) {
+            this.broker = broker;
+            return this;
+        }
+
+        public MqttConfig setClientId(String clientId) {
+            this.clientId = clientId;
+            return this;
+        }
+
+        public MqttConfig setUsername(String username) {
+            this.username = username;
+            return this;
+        }
+
+        public MqttConfig setPassword(String password) {
+            this.password = password;
+            return this;
+        }
+
+        public MqttConfig setConnectionTimeout(int connectionTimeout) {
+            this.connectionTimeout = connectionTimeout;
+            return this;
+        }
+
+        public MqttConfig setKeepAliveInterval(int keepAliveInterval) {
+            this.keepAliveInterval = keepAliveInterval;
+            return this;
+        }
+
+        public MqttConfig setCleanSession(boolean cleanSession) {
+            this.cleanSession = cleanSession;
+            return this;
+        }
+
+        public MqttConfig setAutomaticReconnect(boolean automaticReconnect) {
+            this.automaticReconnect = automaticReconnect;
+            return this;
+        }
+
+        public MqttConfig setMaxInflight(int maxInflight) {
+            this.maxInflight = maxInflight;
+            return this;
+        }
+
+        public MqttConfig setQos(int qos) {
+            this.qos = qos;
+            return this;
+        }
+
+        public String getBroker() {
+            return broker;
+        }
+
+        public String getClientId() {
+            return clientId;
+        }
+
+        public String getUsername() {
+            return username;
+        }
+
+        public String getPassword() {
+            return password;
+        }
+
+        public int getConnectionTimeout() {
+            return connectionTimeout;
+        }
+
+        public int getKeepAliveInterval() {
+            return keepAliveInterval;
+        }
+
+        public boolean isCleanSession() {
+            return cleanSession;
+        }
+
+        public boolean isAutomaticReconnect() {
+            return automaticReconnect;
+        }
+
+        public int getMaxInflight() {
+            return maxInflight;
+        }
+
+        public int getQos() {
+            return qos;
+        }
+    }
+
+    /**
+     * 创建MQTT工具实例
+     */
+    public MqttUtils() {
+    }
+
+    /**
+     * 连接到MQTT服务器
+     *
+     * @param config 连接配置
+     * @throws MqttException 连接异常
+     */
+    public void connect(MqttConfig config) throws MqttException {
+        connect(config, null, null);
+    }
+
+    /**
+     * 连接到MQTT服务器（带回调）
+     *
+     * @param config       连接配置
+     * @param connCallback 连接回调
+     * @param msgCallback  消息回调
+     * @throws MqttException 连接异常
+     */
+    public void connect(MqttConfig config, ConnectionCallback connCallback, MessageCallback msgCallback)
+            throws MqttException {
+
+        this.connectionCallback = connCallback;
+        this.messageCallback = msgCallback;
+
+        try {
+            // 创建MQTT客户端
+            String brokerUrl = config.getBroker().startsWith("tcp://") ?
+                    config.getBroker() : "tcp://" + config.getBroker();
+
+            mqttClient = new MqttClient(brokerUrl, config.getClientId(), new MemoryPersistence());
+
+            // 设置连接选项
+            connectOptions = new MqttConnectOptions();
+            connectOptions.setCleanSession(config.isCleanSession());
+            connectOptions.setConnectionTimeout(config.getConnectionTimeout());
+            connectOptions.setKeepAliveInterval(config.getKeepAliveInterval());
+            connectOptions.setAutomaticReconnect(config.isAutomaticReconnect());
+            connectOptions.setMaxInflight(config.getMaxInflight());
+
+            // 设置用户名密码（如果有）
+            if (config.getUsername() != null && !config.getUsername().isEmpty()) {
+                connectOptions.setUserName(config.getUsername());
+                if (config.getPassword() != null && !config.getPassword().isEmpty()) {
+                    connectOptions.setPassword(config.getPassword().toCharArray());
+                }
+            }
+
+            // 设置回调
+            mqttClient.setCallback(new MqttCallbackExtended() {
+                @Override
+                public void connectComplete(boolean reconnect, String serverURI) {
+                    isConnected = true;
+                    log.info("MQTT连接成功: " + serverURI + (reconnect ? " (重连)" : ""));
+                    if (connectionCallback != null) {
+                        connectionCallback.onConnected();
+                    }
+                }
+
+                @Override
+                public void connectionLost(Throwable cause) {
+                    isConnected = false;
+                    log.error("MQTT连接断开: " + cause.getMessage());
+                    if (connectionCallback != null) {
+                        connectionCallback.onConnectionLost(cause);
+                    }
+                }
+
+                @Override
+                public void messageArrived(String topic, MqttMessage message) {
+                    String payload = new String(message.getPayload(), StandardCharsets.UTF_8);
+                    log.info("收到消息 - Topic: " + topic + ", QoS: " + message.getQos() +
+                            ", Message: " + payload);
+
+                    if (messageCallback != null) {
+                        messageCallback.onMessageReceived(topic, payload);
+                    }
+                }
+
+                @Override
+                public void deliveryComplete(IMqttDeliveryToken token) {
+                    log.info("消息发送完成");
+                    if (messageCallback != null) {
+                        messageCallback.onMessageDelivered(token);
+                    }
+                }
+            });
+
+            // 连接到服务器
+            log.info("正在连接到: " + brokerUrl);
+            mqttClient.connect(connectOptions);
+
+        } catch (MqttException e) {
+            isConnected = false;
+            log.error("连接失败: " + e.getMessage());
+            if (connectionCallback != null) {
+                connectionCallback.onConnectionFailed(e);
+            }
+            throw e;
+        }
+    }
+
+    /**
+     * 发布消息
+     *
+     * @param topic    主题
+     * @param message  消息内容
+     * @param qos      QoS级别 (0, 1, 2)
+     * @param retained 是否保留消息
+     * @throws MqttException 发布异常
+     */
+    public void publish(String topic, String message, int qos, boolean retained) throws MqttException {
+        if (!isConnected) {
+            throw new MqttException(MqttException.REASON_CODE_CLIENT_NOT_CONNECTED);
+        }
+
+        try {
+            MqttMessage mqttMessage = new MqttMessage(message.getBytes(StandardCharsets.UTF_8));
+            mqttMessage.setQos(qos);
+            mqttMessage.setRetained(retained);
+
+            mqttClient.publish(topic, mqttMessage);
+            log.info("发布成功 - Topic: " + topic + ", QoS: " + qos +
+                    ", Retained: " + retained + ", Message: " + message);
+        } catch (MqttException e) {
+            log.error("发布失败: " + e.getMessage());
+            throw e;
+        }
+    }
+
+    /**
+     * 发布消息（简化版）
+     *
+     * @param topic   主题
+     * @param message 消息内容
+     * @throws MqttException 发布异常
+     */
+    public void publish(String topic, String message) throws MqttException {
+        publish(topic, message, 1, false);
+    }
+
+    /**
+     * 订阅主题
+     *
+     * @param topic 主题
+     * @param qos   QoS级别
+     * @throws MqttException 订阅异常
+     */
+    public void subscribe(String topic, int qos) throws MqttException {
+        if (!isConnected) {
+            throw new MqttException(MqttException.REASON_CODE_CLIENT_NOT_CONNECTED);
+        }
+
+        try {
+            mqttClient.subscribe(topic, qos);
+            subscribedTopics.add(topic);
+            log.info("订阅成功 - Topic: " + topic + ", QoS: " + qos);
+        } catch (MqttException e) {
+            log.error("订阅失败: " + e.getMessage());
+            throw e;
+        }
+    }
+
+    /**
+     * 订阅多个主题
+     *
+     * @param topics 主题数组
+     * @param qos    QoS级别数组
+     * @throws MqttException 订阅异常
+     */
+    public void subscribe(String[] topics, int[] qos) throws MqttException {
+        if (!isConnected) {
+            throw new MqttException(MqttException.REASON_CODE_CLIENT_NOT_CONNECTED);
+        }
+
+        try {
+            mqttClient.subscribe(topics, qos);
+            for (String topic : topics) {
+                subscribedTopics.add(topic);
+            }
+            log.info("批量订阅成功 - 主题数量: " + topics.length);
+        } catch (MqttException e) {
+            log.error("批量订阅失败: " + e.getMessage());
+            throw e;
+        }
+    }
+
+    /**
+     * 取消订阅
+     *
+     * @param topic 主题
+     * @throws MqttException 取消订阅异常
+     */
+    public void unsubscribe(String topic) throws MqttException {
+        if (!isConnected) {
+            throw new MqttException(MqttException.REASON_CODE_CLIENT_NOT_CONNECTED);
+        }
+
+        try {
+            mqttClient.unsubscribe(topic);
+            subscribedTopics.remove(topic);
+            log.info("取消订阅成功 - Topic: " + topic);
+        } catch (MqttException e) {
+            log.error("取消订阅失败: " + e.getMessage());
+            throw e;
+        }
+    }
+
+    /**
+     * 断开连接
+     */
+    public void disconnect() {
+        if (mqttClient != null && isConnected) {
+            try {
+                // 取消所有订阅
+                if (!subscribedTopics.isEmpty()) {
+                    String[] topics = subscribedTopics.toArray(new String[0]);
+                    mqttClient.unsubscribe(topics);
+                    subscribedTopics.clear();
+                }
+
+                // 断开连接
+                mqttClient.disconnect();
+                isConnected = false;
+                log.info("MQTT连接已断开");
+
+                if (connectionCallback != null) {
+                    connectionCallback.onDisconnected();
+                }
+
+            } catch (MqttException e) {
+                log.error("断开连接异常: " + e.getMessage());
+            }
+        }
+    }
+
+    /**
+     * 判断是否已连接
+     *
+     * @return 连接状态
+     */
+    public boolean isConnected() {
+        return isConnected && mqttClient != null && mqttClient.isConnected();
+    }
+
+    /**
+     * 获取客户端ID
+     *
+     * @return 客户端ID
+     */
+    public String getClientId() {
+        return mqttClient != null ? mqttClient.getClientId() : null;
+    }
+
+    /**
+     * 获取服务器URI
+     *
+     * @return 服务器URI
+     */
+    public String getServerURI() {
+        return mqttClient != null ? mqttClient.getServerURI() : null;
+    }
+
+    /**
+     * 获取已订阅的主题列表
+     *
+     * @return 主题列表
+     */
+    public List<String> getSubscribedTopics() {
+        return new ArrayList<>(subscribedTopics);
+    }
+
+    /**
+     * 关闭客户端（释放资源）
+     */
+    public void close() {
+        disconnect();
+        try {
+            if (mqttClient != null) {
+                mqttClient.close();
+                log.info("MQTT客户端已关闭");
+            }
+        } catch (MqttException e) {
+            log.error("关闭客户端异常: " + e.getMessage());
+        }
+    }
+
+    /**
+     * 快速连接并发送消息
+     */
+    public static void quickPublish(String broker, String topic, String username,
+                                    String password, String clientId, String message) throws Exception {
+        MqttUtils mqttUtils = new MqttUtils();
+
+        MqttUtils.MqttConfig config = new MqttUtils.MqttConfig(broker);
+
+        if (StringUtils.isNotBlank(username)) {
+            config.setUsername(username);
+        }
+        if (StringUtils.isNotBlank(password)) {
+            config.setPassword(password);
+        }
+        if (StringUtils.isNotBlank(clientId)) {
+            config.setClientId(clientId);
+        }
+
+        config.setConnectionTimeout(15)
+                .setKeepAliveInterval(30)
+                .setQos(1);
+
+        try {
+            mqttUtils.connect(config);
+            mqttUtils.publish(topic, message);
+            log.info("消息发送成功!");
+        } finally {
+            mqttUtils.disconnect();
+        }
+    }
+
+
+    public static void main(String[] args) {
+
+        String broker = "10.9.0.205:1883";
+        String clientId = "QesP7NB4PpAQhupFxlY";
+        String user = "admin";
+        String pwd = "123456";
+        String topic = "v1/devices/me/telemetry";
+        String message = "{\"temp\": 12.8,\"status\": \"ON\"}";
+
+        try {
+            quickPublish(broker, topic, user, pwd, clientId, message);
+        } catch (Exception e) {
+            e.printStackTrace();
+        }
+
+//        MqttUtils mqttUtils = new MqttUtils();
+//        MqttUtils.MqttConfig config = new MqttUtils.MqttConfig(broker)
+//                .setClientId(clientId)
+//                .setUsername(user)
+//                .setPassword(pwd)
+//                .setConnectionTimeout(15)
+//                .setKeepAliveInterval(30)
+//                .setQos(1);
+//
+//        try {
+//            // 连接
+//            mqttUtils.connect(config);
+//
+//            // 发布消息
+//            mqttUtils.publish(topic, message);
+//
+//        } catch (MqttException e) {
+//            e.printStackTrace();
+//        } finally {
+//            // 断开连接
+//            mqttUtils.disconnect();
+//        }
+    }
+}