凯盛数据推送能源管理

@@ -35,6 +35,60 @@
             <artifactId>spring-boot-starter-test</artifactId>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>com.zaxxer</groupId>
+            <artifactId>HikariCP</artifactId>
+            <version>4.0.3</version>
+        </dependency>
+        <dependency>
+            <groupId>com.zaxxer</groupId>
+            <artifactId>HikariCP-java7</artifactId>
+            <version>2.4.13</version>
+        </dependency>
+        <dependency>
+            <groupId>org.postgresql</groupId>
+            <artifactId>postgresql</artifactId>
+            <version>42.7.3</version>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.httpcomponents</groupId>
+            <artifactId>httpclient</artifactId>
+            <version>4.5.8</version>
+        </dependency>
+        <!--工具类-->
+        <dependency>
+            <groupId>org.apache.commons</groupId>
+            <artifactId>commons-lang3</artifactId>
+            <version>3.6</version>
+        </dependency>
+
+        <!--工具类-->
+        <dependency>
+            <groupId>org.apache.commons</groupId>
+            <artifactId>commons-collections4</artifactId>
+            <version>4.2</version>
+        </dependency>
+
+        <dependency>
+            <groupId>com.alibaba</groupId>
+            <artifactId>fastjson</artifactId>
+            <version>1.2.60</version>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-data-redis</artifactId>
+            <version>2.0.9.RELEASE</version>
+        </dependency>
+        <dependency>
+            <groupId>redis.clients</groupId>
+            <artifactId>jedis</artifactId>
+            <version>2.9.0</version>
+        </dependency>
+        <dependency>
+            <groupId>org.eclipse.paho</groupId>
+            <artifactId>org.eclipse.paho.client.mqttv3</artifactId>
+            <version>1.2.5</version>
+        </dependency>
     </dependencies>
 
     <build>

+package com.iot.scheduler.service;
+
+import com.alibaba.fastjson.JSON;
+import com.iot.scheduler.utils.MqttUtils;
+import com.zaxxer.hikari.HikariConfig;
+import com.zaxxer.hikari.HikariDataSource;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.stereotype.Service;
+import org.springframework.util.CollectionUtils;
+
+import java.sql.*;
+import java.time.LocalDateTime;
+import java.time.format.DateTimeFormatter;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * 凯盛设备
+ */
+@Slf4j
+@Service
+public class KSDeviceReportService {
+
+    @Value("${ks.third.jdbcUrl:jdbc:postgresql://192.168.0.249:5432/iot}")
+    private String jdbcUrl;
+    @Value("${ks.third.jdbcUserName:postgres}")
+    private String jdbcUserName;
+
+    @Value("${ks.third.jdbcPassword:1qaz@WSX}")
+    private String jdbcPassword;
+
+    @Value("${ks.third.selectSql:SELECT * FROM ts_kv_dictionary;}")
+    private String selectSql;
+
+    @Value("${ks.third.selectSql1:SELECT * FROM ts_kv_dictionary;}")
+    private String selectSql1;
+
+    @Value("${ks.third.broker:10.9.0.114:1884}")
+    private String broker;
+    @Value("${ks.third.username:yunpower}")
+    private String username;
+    @Value("${ks.third.password:P1o2w3er}")
+    private String password;
+    @Value("${ks.third.topic:storage0}")
+    private String topic;
+    @Value("${ks.third.clientId:2020672015207559170}")
+    private String clientId;
+    @Value("${ks.third.channelSn:8VAUR9H3}")
+    private String channelSn;
+
+    public void deviceElectricity() {
+        log.info("开始执行凯盛电表设备属性上报任务");
+        try {
+            List<Object> needSyncDataList = initConnectAndSelectData("E");
+            if (CollectionUtils.isEmpty(needSyncDataList)) {
+                log.info("电表没有需要上报的数据，任务结束");
+                return;
+            }
+
+            log.info("开始处理{}条电表设备数据并进行上报", needSyncDataList.size());
+            int successCount = 0;
+            int failCount = 0;
+            List<String> failedDevices = new ArrayList<>();
+            Map<Object, String> map = new HashMap<>();
+            map.put(1,"Ep");
+            map.put(2,"Ubc");
+            map.put(3,"Uc");
+            map.put(4,"P");
+            map.put(5,"Uab");
+            map.put(6,"Ib");
+            map.put(7,"Eq");
+            map.put(8,"F");
+            map.put(9,"Q");
+            map.put(10,"Ua");
+            map.put(11,"S");
+            map.put(12,"Ic");
+            map.put(13,"Ub");
+            map.put(14,"Uca");
+            map.put(15,"PSum");
+            map.put(16,"Ia");
+
+            for (int i = 0; i < needSyncDataList.size(); i++) {
+                Object needSyncData = needSyncDataList.get(i);
+                try {
+                    List<Object> dataList = (ArrayList) needSyncData;
+                    String sn = dataList.get(0).toString();
+
+                    // 1. 获取当前时间
+                    LocalDateTime now = LocalDateTime.now();
+                    // 2. 定义间隔分钟数 (例如 5 分钟)
+                    int intervalMinutes = 5;
+                    // 3. 计算向下取整的时间
+                    // 逻辑：获取当前的分钟数，减去 (分钟数 % 间隔)，秒和纳秒设为 0
+                    LocalDateTime roundedTime = now
+                            .withMinute(now.getMinute() - (now.getMinute() % intervalMinutes))
+                            .withSecond(0)
+                            .withNano(0);
+
+                    // 4. 格式化输出
+                    DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
+
+                    for (int j = 1; j < dataList.size(); j++) {
+                        Map<String, Object> properties = new HashMap<>(5);
+                        Object o = dataList.get(j);
+                        if (o == null) {
+                            continue;
+                        }
+                        String deviceVarSn = "ks_" + sn + map.get(j);
+                        log.info("===deviceVarSn==" + deviceVarSn);
+                        properties.put("channelSn", channelSn);
+                        properties.put("deviceVarSn", deviceVarSn);
+                        properties.put("value", Double.parseDouble(o.toString().trim()));
+                        properties.put("saveTime", roundedTime.format(formatter));
+                        properties.put("collectTime", roundedTime.format(formatter));
+                        try {
+                            log.debug("电表开始MQTT发布: sn={}", sn);
+                            MqttUtils.quickPublish(broker, topic, username, password, clientId, JSON.toJSONString(properties));
+                            successCount++;
+
+                        } catch (Exception e) {
+                            failCount++;
+                            failedDevices.add(sn);
+                            log.error("电表设备上报失败: sn={}, clientId={}, topic={}, 错误信息: {}",
+                                    sn, clientId, topic, e.getMessage());
+                            log.error("电表详细异常信息:", e);
+                        }
+
+                        // 可选：添加延迟，避免发送过快
+                        try {
+                            Thread.sleep(10); // 10毫秒延迟
+                        } catch (InterruptedException e) {
+                            log.warn("电表线程休眠被中断", e);
+                            Thread.currentThread().interrupt();
+                        }
+                    }
+                } catch (Exception e) {
+                    failCount++;
+                    log.error("电表处理第{}条数据时发生异常，数据内容: {}", i + 1, needSyncData, e);
+                }
+            }
+
+            log.info("sn电表设备属性上报任务完成，成功: {}条，失败: {}条，总计: {}条",
+                    successCount, failCount, needSyncDataList.size());
+
+        } catch (Exception e) {
+            log.error("sn执行电表设备属性上报任务时发生异常", e);
+        } finally {
+            log.info("凯盛电表设备属性上报任务执行结束");
+        }
+    }
+
+    public void deviceWater() {
+        log.info("开始执行凯盛水表设备属性上报任务");
+        try {
+            List<Object> needSyncDataList = initConnectAndSelectData("W");
+            if (CollectionUtils.isEmpty(needSyncDataList)) {
+                log.info("水表没有需要上报的数据，任务结束");
+                return;
+            }
+
+            log.info("开始处理{}条水表设备数据并进行上报", needSyncDataList.size());
+            int successCount = 0;
+            int failCount = 0;
+            List<String> failedDevices = new ArrayList<>();
+            Map<Object, String> map = new HashMap<>();
+            map.put(1,"devid");
+            map.put(2,"zll");
+            map.put(3,"ssll");
+
+
+            for (int i = 0; i < needSyncDataList.size(); i++) {
+                Object needSyncData = needSyncDataList.get(i);
+                try {
+                    List<Object> dataList = (ArrayList) needSyncData;
+                    String sn = dataList.get(0).toString();
+
+                    // 1. 获取当前时间
+                    LocalDateTime now = LocalDateTime.now();
+                    // 2. 定义间隔分钟数 (例如 5 分钟)
+                    int intervalMinutes = 5;
+                    // 3. 计算向下取整的时间
+                    // 逻辑：获取当前的分钟数，减去 (分钟数 % 间隔)，秒和纳秒设为 0
+                    LocalDateTime roundedTime = now
+                            .withMinute(now.getMinute() - (now.getMinute() % intervalMinutes))
+                            .withSecond(0)
+                            .withNano(0);
+
+                    // 4. 格式化输出
+                    DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
+
+                    for (int j = 1; j < dataList.size(); j++) {
+                        Map<String, Object> properties = new HashMap<>(5);
+                        Object o = dataList.get(j).toString();
+                        if (o == null) {
+                            continue;
+                        }
+                        String deviceVarSn = "ks_" + sn + map.get(j);
+                        log.info("===deviceVarSn==" + deviceVarSn);
+                        properties.put("channelSn", channelSn);
+                        properties.put("deviceVarSn", deviceVarSn);
+                        properties.put("value", Double.parseDouble(o.toString().trim()));
+                        properties.put("saveTime", roundedTime.format(formatter));
+                        properties.put("collectTime", roundedTime.format(formatter));
+                        try {
+                            log.debug("水表开始MQTT发布: sn={}", sn);
+                            MqttUtils.quickPublish(broker, topic, username, password, clientId, JSON.toJSONString(properties));
+                            successCount++;
+
+                        } catch (Exception e) {
+                            failCount++;
+                            failedDevices.add(sn);
+                            log.error("水表设备上报失败: sn={}, clientId={}, topic={}, 错误信息: {}",
+                                    sn, clientId, topic, e.getMessage());
+                            log.error("水表详细异常信息:", e);
+                        }
+
+                        // 可选：添加延迟，避免发送过快
+                        try {
+                            Thread.sleep(10); // 10毫秒延迟
+                        } catch (InterruptedException e) {
+                            log.warn("水表线程休眠被中断", e);
+                            Thread.currentThread().interrupt();
+                        }
+                    }
+                } catch (Exception e) {
+                    failCount++;
+                    log.error("水表处理第{}条数据时发生异常，数据内容: {}", i + 1, needSyncData, e);
+                }
+            }
+
+            log.info("水表设备属性上报任务完成，成功: {}条，失败: {}条，总计: {}条",
+                    successCount, failCount, needSyncDataList.size());
+
+        } catch (Exception e) {
+            log.error("水表执行设备属性上报任务时发生异常", e);
+        } finally {
+            log.info("凯盛水表设备属性上报任务执行结束");
+        }
+    }
+
+    private List<Object> initConnectAndSelectData(String type) {
+        Connection connection = null;
+        PreparedStatement statement = null;
+        ResultSet resultSet = null;
+        HikariDataSource dataSource = null;
+        List<Object> resultList = new ArrayList<>();
+
+        log.info("开始连接数据库，URL: {}", jdbcUrl);
+
+        try {
+            HikariConfig config = new HikariConfig();
+            config.setJdbcUrl(jdbcUrl);
+            config.setUsername(jdbcUserName);
+            config.setPassword(jdbcPassword);
+            config.setDriverClassName("org.postgresql.Driver");
+            config.setMaximumPoolSize(5);
+            config.setMinimumIdle(5);
+            config.setConnectionTimeout(60000);
+            config.setConnectionTestQuery("SELECT 1");
+
+            dataSource = new HikariDataSource(config);
+            log.info("Hikari连接池配置完成");
+
+            connection = dataSource.getConnection();
+            log.info("数据库连接成功");
+            if ("E".equals(type)) {
+                statement = connection.prepareStatement(selectSql);
+                log.info("执行SQL查询: {}", selectSql);
+            } else {
+                statement = connection.prepareStatement(selectSql1);
+                log.info("执行SQL查询: {}", selectSql1);
+            }
+
+            resultSet = statement.executeQuery();
+            ResultSetMetaData metaData = resultSet.getMetaData();
+            int columnCount = metaData.getColumnCount();
+            log.info("查询结果集元数据获取成功，共{}列", columnCount);
+
+            int rowCount = 0;
+            while (resultSet.next()) {
+                List<Object> result = new ArrayList<>(columnCount);
+                for (int index = 1; index <= columnCount; index++) {
+                    int columnType = metaData.getColumnType(index);
+                    Object value = getTypedValue(resultSet, index, columnType);
+                    result.add(value);
+                }
+                resultList.add(result);
+                rowCount++;
+
+                // 每处理1000行记录一次日志
+                if (rowCount % 1000 == 0) {
+                    log.info("已处理{}行数据", rowCount);
+                }
+            }
+
+            log.info("数据查询完成，共获取{}行数据", rowCount);
+
+        } catch (SQLException e) {
+            log.error("数据库操作异常，URL: {}, 用户名: {}", jdbcUrl, jdbcUserName, e);
+        } catch (Exception e) {
+            log.error("初始化数据库连接或查询数据时发生异常", e);
+        } finally {
+            // 释放资源
+            try {
+                if (resultSet != null) resultSet.close();
+                if (statement != null) statement.close();
+                if (connection != null) connection.close();
+                log.info("数据库连接资源已释放");
+            } catch (SQLException e) {
+                log.error("关闭数据库资源时发生异常", e);
+            }
+
+            if (dataSource != null) {
+                try {
+                    dataSource.close();
+                    log.info("HikariDataSource连接池已关闭");
+                } catch (Exception e) {
+                    log.error("关闭HikariDataSource连接池时发生异常", e);
+                }
+            }
+        }
+
+        log.info("数据库操作完成，返回{}条记录", resultList.size());
+        return resultList;
+    }
+
+    private Object getTypedValue(ResultSet rs, int index, int sqlType) throws SQLException {
+        Object value;
+
+        try {
+            switch (sqlType) {
+                case Types.BIT:
+                case Types.BOOLEAN:
+                    value = rs.getBoolean(index);
+                    return rs.wasNull() ? null : value;
+
+                case Types.TINYINT:
+                case Types.SMALLINT:
+                case Types.INTEGER:
+                    value = rs.getInt(index);
+                    return rs.wasNull() ? null : value;
+
+                case Types.BIGINT:
+                    value = rs.getLong(index);
+                    return rs.wasNull() ? null : value;
+
+                case Types.FLOAT:
+                case Types.REAL:
+                    value = rs.getFloat(index);
+                    return rs.wasNull() ? null : value;
+
+                case Types.DOUBLE:
+                    value = rs.getDouble(index);
+                    return rs.wasNull() ? null : value;
+
+                case Types.NUMERIC:
+                case Types.DECIMAL:
+                    value = rs.getBigDecimal(index);
+                    return rs.wasNull() ? null : value;
+
+                case Types.CHAR:
+                case Types.VARCHAR:
+                case Types.LONGVARCHAR:
+                case Types.NCHAR:
+                case Types.NVARCHAR:
+                case Types.LONGNVARCHAR:
+                    value = rs.getString(index);
+                    return rs.wasNull() ? null : value;
+
+                case Types.DATE:
+                    Date date = rs.getDate(index);
+                    return date != null ? date.toLocalDate() : null;
+
+                case Types.TIME:
+                    Time time = rs.getTime(index);
+                    return time != null ? time.toLocalTime() : null;
+
+                case Types.TIMESTAMP:
+                    Timestamp timestamp = rs.getTimestamp(index);
+                    return timestamp != null ? timestamp.toLocalDateTime() : null;
+
+                case Types.BINARY:
+                case Types.VARBINARY:
+                case Types.LONGVARBINARY:
+                    value = rs.getBytes(index);
+                    return rs.wasNull() ? null : value;
+
+                case Types.BLOB:
+                    value = rs.getBlob(index);
+                    return rs.wasNull() ? null : value;
+
+                case Types.CLOB:
+                    value = rs.getClob(index);
+                    return rs.wasNull() ? null : value;
+
+                default:
+                    value = rs.getObject(index);
+                    return rs.wasNull() ? null : value;
+            }
+        } catch (SQLException e) {
+            log.error("获取结果集第{}列数据时发生异常，数据类型: {}", index, sqlType, e);
+            throw e;
+        }
+    }
+}

+package com.iot.scheduler.utils;
+
+import io.micrometer.common.util.StringUtils;
+import lombok.extern.slf4j.Slf4j;
+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();
+
+        MqttConfig config = new 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();
+//        }
+    }
+}

+package com.iot.scheduler.utils;
+
+import lombok.extern.slf4j.Slf4j;
+
+import java.sql.*;
+
+@Slf4j
+public class SqlTypedValueUtils {
+
+    public static Object getTypedValue(ResultSet rs, int index, int sqlType) throws SQLException {
+        Object value;
+
+        try {
+            switch (sqlType) {
+                case Types.BIT:
+                case Types.BOOLEAN:
+                    value = rs.getBoolean(index);
+                    return rs.wasNull() ? null : value;
+
+                case Types.TINYINT:
+                case Types.SMALLINT:
+                case Types.INTEGER:
+                    value = rs.getInt(index);
+                    return rs.wasNull() ? null : value;
+
+                case Types.BIGINT:
+                    value = rs.getLong(index);
+                    return rs.wasNull() ? null : value;
+
+                case Types.FLOAT:
+                case Types.REAL:
+                    value = rs.getFloat(index);
+                    return rs.wasNull() ? null : value;
+
+                case Types.DOUBLE:
+                    value = rs.getDouble(index);
+                    return rs.wasNull() ? null : value;
+
+                case Types.NUMERIC:
+                case Types.DECIMAL:
+                    value = rs.getBigDecimal(index);
+                    return rs.wasNull() ? null : value;
+
+                case Types.CHAR:
+                case Types.VARCHAR:
+                case Types.LONGVARCHAR:
+                case Types.NCHAR:
+                case Types.NVARCHAR:
+                case Types.LONGNVARCHAR:
+                    value = rs.getString(index);
+                    return rs.wasNull() ? null : value;
+
+                case Types.DATE:
+                    Date date = rs.getDate(index);
+                    return date != null ? date.toLocalDate() : null;
+
+                case Types.TIME:
+                    Time time = rs.getTime(index);
+                    return time != null ? time.toLocalTime() : null;
+
+                case Types.TIMESTAMP:
+                    Timestamp timestamp = rs.getTimestamp(index);
+                    return timestamp != null ? timestamp.toLocalDateTime() : null;
+
+                case Types.BINARY:
+                case Types.VARBINARY:
+                case Types.LONGVARBINARY:
+                    value = rs.getBytes(index);
+                    return rs.wasNull() ? null : value;
+
+                case Types.BLOB:
+                    value = rs.getBlob(index);
+                    return rs.wasNull() ? null : value;
+
+                case Types.CLOB:
+                    value = rs.getClob(index);
+                    return rs.wasNull() ? null : value;
+
+                default:
+                    value = rs.getObject(index);
+                    return rs.wasNull() ? null : value;
+            }
+        } catch (SQLException e) {
+            log.error("获取结果集第{}列数据时发生异常，数据类型: {}", index, sqlType, e);
+            throw e;
+        }
+    }
+}

+package com.iot.scheduler.zone;
+
+import com.iot.scheduler.service.KSDeviceReportService;
+import com.iot.scheduler.task.AbstractZoneScheduler;
+import jakarta.annotation.Resource;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.scheduling.annotation.Scheduled;
+import org.springframework.stereotype.Component;
+
+@Slf4j
+@Component
+public class KaiShengZoneScheduler extends AbstractZoneScheduler {
+    @Resource
+    private KSDeviceReportService ksDeviceReportService;
+
+    @Override
+    protected String getZoneName() {
+        return "KaiSheng (凯盛)";
+    }
+
+    @Scheduled(cron = "${scheduler.ks.push:0 0/5 * * * ?}")
+    public void pushKsDevicesToThirdParty() {
+        String taskName = "ks Push Devices (IoT -> 3rd Party)";
+        logStart(taskName);
+        try {
+            log.info("[{}] ks pushing devices...", getZoneName());
+            ksDeviceReportService.deviceElectricity();
+//            ksDeviceReportService.deviceWater();
+            Thread.sleep(1000);
+        } catch (Exception e) {
+            logError(taskName, e);
+        } finally {
+            logEnd(taskName);
+        }
+    }
+}

@@ -5,7 +5,7 @@ spring:
     banner-mode: off
 
 server:
-  port: 8080
+  port: 32211
 
 logging:
   level:
@@ -24,3 +24,83 @@ scheduler:
   panji:
     pull: "0 0/5 * * * ?"
     push: "0 0/10 * * * ?"
+  ks:
+    pull: "0 0/5 * * * ?"
+    push: "0 0/5 * * * ?"
+ks:
+  third:
+    jdbcUrl: "jdbc:postgresql://192.168.0.249:5432/iot"
+    jdbcUserName: "postgres"
+    jdbcPassword: "1qaz@WSX"
+    selectSql: "SELECT
+                  de.sn AS sn,
+                  tkl1.dbl_v AS Ep,
+                  tkl2.dbl_v AS Ubc,
+                  tkl3.dbl_v AS Uc,
+                  tkl4.dbl_v AS P,
+                  tkl5.dbl_v AS Uab,
+                  tkl6.dbl_v AS Ib,
+                  tkl7.dbl_v AS Eq,
+                  tkl8.dbl_v AS F,
+                  tkl9.dbl_v AS Q,
+                  tkl10.dbl_v AS Ua,
+                  tkl11.dbl_v AS S,
+                  tkl12.dbl_v AS Ic,
+                  tkl13.dbl_v AS Ub,
+                  tkl14.dbl_v AS Uca,
+                  tkl15.dbl_v AS PSum,
+                  tkl16.dbl_v AS Ia
+                FROM
+                  device de
+                  LEFT JOIN ts_kv_latest tkl1 ON de.id = tkl1.entity_id
+                  AND tkl1.key = 64
+                  LEFT JOIN ts_kv_latest tkl2 ON de.id = tkl2.entity_id
+                  AND tkl2.key = 65
+                  LEFT JOIN ts_kv_latest tkl3 ON de.id = tkl3.entity_id
+                  AND tkl3.key = 66
+                  LEFT JOIN ts_kv_latest tkl4 ON de.id = tkl4.entity_id
+                  AND tkl4.key = 67
+                  LEFT JOIN ts_kv_latest tkl5 ON de.id = tkl5.entity_id
+                  AND tkl5.key = 68
+                  LEFT JOIN ts_kv_latest tkl6 ON de.id = tkl6.entity_id
+                  AND tkl6.key = 69
+                  LEFT JOIN ts_kv_latest tkl7 ON de.id = tkl7.entity_id
+                  AND tkl7.key = 70
+                  LEFT JOIN ts_kv_latest tkl8 ON de.id = tkl8.entity_id
+                  AND tkl8.key = 71
+                  LEFT JOIN ts_kv_latest tkl9 ON de.id = tkl9.entity_id
+                  AND tkl9.key = 72
+                  LEFT JOIN ts_kv_latest tkl10 ON de.id = tkl10.entity_id
+                  AND tkl10.key = 73
+                  LEFT JOIN ts_kv_latest tkl11 ON de.id = tkl11.entity_id
+                  AND tkl11.key = 74
+                  LEFT JOIN ts_kv_latest tkl12 ON de.id = tkl12.entity_id
+                  AND tkl12.key = 75
+                  LEFT JOIN ts_kv_latest tkl13 ON de.id = tkl13.entity_id
+                  AND tkl13.key = 76
+                  LEFT JOIN ts_kv_latest tkl14 ON de.id = tkl14.entity_id
+                  AND tkl14.key = 77
+                  LEFT JOIN ts_kv_latest tkl15 ON de.id = tkl15.entity_id
+                  AND tkl15.key = 78
+                  LEFT JOIN ts_kv_latest tkl16 ON de.id = tkl16.entity_id
+                  AND tkl16.key = 79
+                  WHERE
+                  de.organization_id='1697500a-dc11-45cc-88f5-2ad47472a9bb'
+                  AND de.device_profile_id = 'c2401630-ffec-11f0-926f-2f3182abc65f'"
+
+    selectSql1: "SELECT
+                de.sn AS sn,
+                tkl1.dbl_v AS devid,
+                tkl2.dbl_v AS zll,
+                tkl3.dbl_v AS ssll
+              FROM
+                device de
+                LEFT JOIN ts_kv_latest tkl1 ON de.id = tkl1.entity_id
+                AND tkl1.key = 58
+                LEFT JOIN ts_kv_latest tkl2 ON de.id = tkl2.entity_id
+                AND tkl2.key = 83
+                LEFT JOIN ts_kv_latest tkl3 ON de.id = tkl3.entity_id
+                AND tkl3.key = 84
+                WHERE
+                de.organization_id='1697500a-dc11-45cc-88f5-2ad47472a9bb'
+                AND de.device_profile_id = '4e404b10-ffe7-11f0-926f-2f3182abc65f'"