凯盛数据推送能源管理

@@ -117,7 +117,7 @@ public class KSDeviceReportService {
                         properties.put("collectTime", roundedTime.format(formatter));
                         try {
                             log.debug("电表开始MQTT发布: sn={}", sn);
-                            MqttUtils.quickPublish(broker, topic, username, password, clientId, JSON.toJSONString(properties));
+                            MqttUtils.publish(broker, topic, username, password, clientId, JSON.toJSONString(properties));
                             successCount++;
 
                         } catch (Exception e) {
@@ -206,7 +206,7 @@ public class KSDeviceReportService {
                         properties.put("collectTime", roundedTime.format(formatter));
                         try {
                             log.debug("水表开始MQTT发布: sn={}", sn);
-                            MqttUtils.quickPublish(broker, topic, username, password, clientId, JSON.toJSONString(properties));
+                            MqttUtils.publish(broker, topic, username, password, clientId, JSON.toJSONString(properties));
                             successCount++;
 
                         } catch (Exception e) {

@@ -481,6 +481,32 @@ public class MqttUtils {
         }
     }
 
+    public static void publish(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)
+                .setClientId(clientId)
+                .setUsername(username)
+                .setPassword(password)
+                .setConnectionTimeout(15)
+                .setKeepAliveInterval(30)
+                .setQos(1);
+
+        try {
+            // 连接
+            mqttUtils.connect(config);
+
+            // 发布消息
+            mqttUtils.publish(topic, message);
+
+        } catch (MqttException e) {
+            e.printStackTrace();
+        } finally {
+            // 断开连接
+            mqttUtils.disconnect();
+        }
+    }
+
 
     public static void main(String[] args) {
 

@@ -18,14 +18,14 @@ public class KaiShengZoneScheduler extends AbstractZoneScheduler {
         return "KaiSheng (凯盛)";
     }
 
-    @Scheduled(cron = "${scheduler.ks.push:0 0/2 * * * ?}")
+    @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();
+            ksDeviceReportService.deviceWater();
             Thread.sleep(1000);
         } catch (Exception e) {
             logError(taskName, e);

@@ -25,8 +25,8 @@ scheduler:
     pull: "0 0/5 * * * ?"
     push: "0 0/10 * * * ?"
   ks:
-    pull: "0 0/2 * * * ?"
-    push: "0 0/2 * * * ?"
+    pull: "0 0/5 * * * ?"
+    push: "0 0/5 * * * ?"
 ks:
   third:
     jdbcUrl: "jdbc:postgresql://192.168.0.249:5432/iot"