mqtt初始化链接调试

.idea/compiler.xml

@@ -0,0 +1,19 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project version="4">
+  <component name="CompilerConfiguration">
+    <annotationProcessing>
+      <profile default="true" name="Default" enabled="true" />
+      <profile name="Maven default annotation processors profile" enabled="true">
+        <sourceOutputDir name="target/generated-sources/annotations" />
+        <sourceTestOutputDir name="target/generated-test-sources/test-annotations" />
+        <outputRelativeToContentRoot value="true" />
+        <module name="springboot" />
+      </profile>
+    </annotationProcessing>
+  </component>
+  <component name="JavacSettings">
+    <option name="ADDITIONAL_OPTIONS_OVERRIDE">
+      <module name="springboot" options="-parameters" />
+    </option>
+  </component>
+</project>

.idea/encodings.xml

@@ -0,0 +1,6 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project version="4">
+  <component name="Encoding">
+    <file url="file://$PROJECT_DIR$/src/main/java" charset="UTF-8" />
+  </component>
+</project>

.idea/jarRepositories.xml

@@ -0,0 +1,20 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project version="4">
+  <component name="RemoteRepositoriesConfiguration">
+    <remote-repository>
+      <option name="id" value="central" />
+      <option name="name" value="Maven Central repository" />
+      <option name="url" value="https://repo1.maven.org/maven2" />
+    </remote-repository>
+    <remote-repository>
+      <option name="id" value="jboss.community" />
+      <option name="name" value="JBoss Community repository" />
+      <option name="url" value="https://repository.jboss.org/nexus/content/repositories/public/" />
+    </remote-repository>
+    <remote-repository>
+      <option name="id" value="central" />
+      <option name="name" value="Central Repository" />
+      <option name="url" value="https://maven.aliyun.com/repository/central" />
+    </remote-repository>
+  </component>
+</project>

API接口列表.csv

@@ -0,0 +1,51 @@
+功能名称,功能描述,接口路径
+服务器状态检查,检查服务器是否正常运行,/check/server.do
+UDP测试发送,测试UDP数据包发送功能,/check/testSendUdp.do
+管网专题页点位查询（单点）,根据主键查询管网专题页点位详细信息,/pipeTopicMPoint/selectById.do
+管网专题页点位新增,新增管网专题页点位信息,/pipeTopicMPoint/save.do
+管网专题页点位更新,更新管网专题页点位信息,/pipeTopicMPoint/update.do
+管网专题页点位删除（单点）,根据主键删除单个管网专题页点位,/pipeTopicMPoint/deleteById.do
+管网专题页点位批量删除,批量删除多个管网专题页点位,/pipeTopicMPoint/deleteBatch.do
+管网专题页点位分页查询,分页查询管网专题页点位列表,/pipeTopicMPoint/list.do
+管网专题页数据获取,获取管网专题页展示数据,/pipeTopicMPoint/getData.do
+管道设备查询（单点）,根据主键查询管道设备详细信息,/pipeline-equipment/selectById.do
+管道设备删除（单点）,删除指定主键的管道设备信息,/pipeline-equipment/deleteById.do
+管道设备删除（单点）,删除指定主键的管道设备信息（重复接口）,/pipeline-equipment/deleteByIds.do
+管道设备新增,新增管道设备信息,/pipeline-equipment/save.do
+管道设备更新,更新管道设备信息,/pipeline-equipment/update.do
+管道设备分页查询,分页查询管道设备列表,/pipeline-equipment/list.do
+管道设备类型获取,获取所有管道设备类型,/pipeline-equipment/device-types
+管道设备附属点位查询（单点）,根据主键查询管道设备与测点关联信息,/pipeline-equipment-mpoint/selectById.do
+管道设备附属点位删除（单点）,删除指定主键的管道设备与测点关联信息,/pipeline-equipment-mpoint/deleteById.do
+管道设备附属点位新增,新增管道设备与测点关联信息,/pipeline-equipment-mpoint/save.do
+管道设备附属点位更新,更新管道设备与测点关联信息,/pipeline-equipment-mpoint/update.do
+管道设备附属点位分页查询,分页查询管道设备与测点关联信息列表,/pipeline-equipment-mpoint/list.do
+管道设备附属点位一键导入,根据设备类型一键导入点位信息,/pipeline-equipment-mpoint/import4Type.do
+管网测点库查询（单点）,根据主键查询管网测点库记录,/pipeline-mpoint-library/selectById.do
+管网测点库删除（单点）,删除指定主键的管网测点库记录,/pipeline-mpoint-library/deleteById.do
+管网测点库批量删除,批量删除多个管网测点库记录,/pipeline-mpoint-library/deleteByIds.do
+管网测点库新增,新增管网测点库记录,/pipeline-mpoint-library/save.do
+管网测点库更新,更新管网测点库记录,/pipeline-mpoint-library/update.do
+管网测点库分页查询,分页查询管网测点库列表,/pipeline-mpoint-library/list.do
+Kafka消息发送测试,测试Kafka消息发送功能,/api/kafka/testSendMessage
+MQTT配置查询（单点）,根据主键查询MQTT配置信息,/mqttConfig/getById
+MQTT连接启动,启动MQTT连接,/mqttConfig/connectStart.do
+MQTT连接建立,建立MQTT连接,/mqttConfig/connect.do
+MQTT配置主题新增,新增MQTT配置主题信息,/mqttConfigTopic/save.do
+MQTT配置主题删除,删除MQTT配置主题信息,/mqttConfigTopic/delete.do
+MQTT客户端状态查询（单点）,查询指定网关是否在线（单个网关）,/mqtt/getClientStatus.do
+MQTT客户端列表状态查询,返回所有在线的网关（所有网关）,/mqtt/getClientListStatus.do
+MQTT客户端列表状态查询（按厂）,根据厂ID返回该厂所有网关在线状态,/mqtt/getClientListStatus4UnitId.do
+OPC UA节点值读取,读取单个OPC UA节点的值,/api/opcua/read
+OPC UA节点值写入,写入单个OPC UA节点的值,/api/opcua/write
+OPC UA节点订阅取消,取消OPC UA节点订阅,/api/opcua/unsubscribeFromNode
+OPC UA订阅同步,同步OPC UA订阅点位,/api/opcua/sysSub
+RabbitMQ消息发送,发送RabbitMQ消息到指定队列,/mq/sendMQ
+测点更新,更新测点信息,/mPoint/update.do
+测点查询（单点）,根据ID查询测点信息,/mPoint/selectById.do
+测点实时数据获取（多点位）,管网平台获取实时数据接口（支持多个点位）,/mPoint/getDataByMpCode.do
+公司信息新增,新增公司信息,/company/save.do
+公司信息删除,删除公司信息,/company/delete.do
+用户测试接口1,用户测试接口（返回成功）,/user/test1
+用户测试接口2,用户测试接口（返回失败）,/user/test2
+设备上下线消息处理,通过EMQX WebHook获取设备上下线消息并处理,/webHook/onJkWebHook

src/main/java/com/sipai/entity/scada/MPointHistory.java

@@ -3,6 +3,7 @@ package com.sipai.entity.scada;
 import com.sipai.entity.base.SQLAdapter;
 import java.io.Serializable;
 import java.math.BigDecimal;
+import java.util.Date;
 
 public class MPointHistory extends SQLAdapter implements Serializable{
 	/**
@@ -151,4 +152,24 @@ public class MPointHistory extends SQLAdapter implements Serializable{
 		return serialVersionUID;
 	}
 
+    @Override
+    public String toString() {
+        final StringBuilder sb = new StringBuilder("MPointHistory{");
+        sb.append("itemid=").append(itemid);
+        sb.append(", parmvalue=").append(parmvalue);
+        sb.append(", measuredt='").append(measuredt).append('\'');
+        sb.append(", memotype='").append(memotype).append('\'');
+        sb.append(", memo='").append(memo).append('\'');
+        sb.append(", userid='").append(userid).append('\'');
+        sb.append(", insdt='").append(insdt).append('\'');
+        sb.append(", tbName='").append(tbName).append('\'');
+        sb.append(", tag=").append(tag);
+        sb.append(", year='").append(year).append('\'');
+        sb.append(", month='").append(month).append('\'');
+        sb.append(", day='").append(day).append('\'');
+        sb.append(", hour='").append(hour).append('\'');
+        sb.append(", min='").append(min).append('\'');
+        sb.append('}');
+        return sb.toString();
+    }
 }

src/main/java/com/sipai/schedule/DataSynJob.java

@@ -47,21 +47,22 @@ public class DataSynJob {
     }
 
     @Async
-//    @Scheduled(cron = "0 0/3 * * * ?")//数据转发
+    @Scheduled(cron = "0 0 0/1 * * ?")//数据转发
     public void job2() {
         if (!scheduledEnabled) return;  // 手动拦截
         String addstr = "zhuanfa";
         System.out.println("开始定时器-----------------" + CommUtil.nowDate() + "-----------------" + addstr);
-        List<MPoint> list2 = this.mPointService.selectListByWhere("where bizid = '0791CNWS' and SignalType='AI' and MeasureDT>DATE_SUB(NOW(), INTERVAL 5 MINUTE) and source_type='auto' ");
+//        List<MPoint> list2 = this.mPointService.selectListByWhere("where bizid = '0533JS' and SignalType='AI' and MeasureDT>DATE_SUB(NOW(), INTERVAL 5 MINUTE) and source_type='auto' ");
-        if (list2 != null && list2.size() > 0) {
+        List<MPoint> list2 = this.mPointService.selectListByWhere("where bizid = '0533JS' and SignalType='AI' and source_type='auto' ");
-            for (int j = 0; j < list2.size(); ++j) {
+        if (list2 != null && !list2.isEmpty()) {
+            for (MPoint mPoint : list2) {
                 MPointHistory mPointHistory = new MPointHistory();
-                mPointHistory.setParmvalue(list2.get(j).getParmvalue());
+                mPointHistory.setParmvalue(mPoint.getParmvalue());
-                mPointHistory.setMeasuredt(list2.get(j).getMeasuredt());
+                mPointHistory.setMeasuredt(mPoint.getMeasuredt());
-                mPointHistory.setTbName(list2.get(j).getMpointcode());
+                mPointHistory.setTbName(mPoint.getMpointcode());
                 mPointHistory.setUserid("data_job");
                 mPointHistory.setInsdt(CommUtil.nowDate());
-                mPointHistoryService.saveByCreate(list2.get(j).getBizid(), mPointHistory);
+                mPointHistoryService.saveByCreate(mPoint.getBizid(), mPointHistory);
             }
             System.out.println("完成一次（有数据）转发，执行了" + list2.size() + "次点");
         } else {

src/main/java/com/sipai/service/mqtt/PushCallback.java

@@ -127,7 +127,7 @@ public class PushCallback implements MqttCallbackExtended {
     public void messageArrived(String topic, MqttMessage message) {
         try {
             JSONArray jsonArray = JSONArray.parseArray(message.toString());
-//            System.out.println(topic + "===开始===" + CommUtil.nowDate() + "===" + jsonArray.size());
+            System.out.println(topic + "===开始===" + CommUtil.nowDate() + "===" + jsonArray.size());
             String unitId = topic.substring(0, 4);//截取topic的前面部分作为厂id
             MqttService mqttService = (MqttService) SpringContextUtil.getBean("mqttService");
             //正常的主题 -- 数据处理

src/main/java/com/sipai/service/mqtt/impl/MqttConfigServiceImpl.java

@@ -12,6 +12,7 @@ import com.sipai.entity.mqtt.MqttConfigTopic;
 import com.sipai.service.mqtt.MqttConfigService;
 import com.sipai.service.mqtt.MqttConfigTopicService;
 import com.sipai.tools.CommUtil;
+import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.codec.binary.Base64;
 import org.eclipse.paho.client.mqttv3.MqttClient;
 import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
@@ -37,6 +38,7 @@ import java.util.List;
  * @Date: 2021/03/24/16:45
  * @Description:
  */
+@Slf4j
 @Service
 public class MqttConfigServiceImpl implements MqttConfigService {
     private final MqttProperties mqttProperties;
@@ -152,6 +154,7 @@ public class MqttConfigServiceImpl implements MqttConfigService {
                     mqttClient1.setCallback(new PushCallback(mqttClient1, connOpts, topic, qos));
                     //订阅
                     mqttClient1.subscribe(topic, qos);
+                    System.out.println("已连接成功");
                 } else {
                     System.out.println("重新连接");
                     //连接
@@ -160,12 +163,15 @@ public class MqttConfigServiceImpl implements MqttConfigService {
                     mqttClient1.setCallback(new PushCallback(mqttClient1, connOpts, topic, qos));
                     //订阅
                     mqttClient1.subscribe(topic, qos);
+                    System.out.println("重新连接成功");
                 }
 
             } catch (MqttException me) {
+                log.error("MqttException:",me);
                 me.printStackTrace();
             }
         }
+        System.out.println(clientId);
         return clientId;
     }
 

src/main/java/com/sipai/service/mqtt/impl/MqttServiceImpl.java

@@ -2,6 +2,7 @@ package com.sipai.service.mqtt.impl;
 
 import com.alibaba.fastjson.JSONArray;
 import com.alibaba.fastjson.JSONObject;
+import com.sipai.dao.scada.MPointHistoryDao;
 import com.sipai.entity.mqtt.Mqtt;
 import com.sipai.entity.mqtt.MqttProperties;
 import com.sipai.entity.scada.MPoint;
@@ -29,10 +30,9 @@ import org.springframework.stereotype.Service;
 
 import java.math.BigDecimal;
 import java.math.RoundingMode;
-import java.util.ArrayList;
+import java.time.LocalDateTime;
-import java.util.List;
+import java.time.format.DateTimeFormatter;
-import java.util.Map;
+import java.util.*;
-import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.TimeUnit;
 
@@ -49,6 +49,10 @@ public class MqttServiceImpl implements MqttService {
     private ListenerPointService listenerPointService;
     @Autowired
     private RedissonClient redissonClient;
+    @Autowired
+    private MPointHistoryService mPointHistoryService;
+    @Autowired
+    MPointHistoryDao mPointHistoryDao;
 
     private static MqttClient mqttClient;
     private static String ipStr = "";
@@ -90,10 +94,13 @@ public class MqttServiceImpl implements MqttService {
         return mqttClient;
     }
 
+    DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
+
     @Async("getAsyncMqttHandle")
     @Override
     public void doHandle(String bizId, String topic, String ip4, String port, JSONArray jsonArray) {
-//        System.out.println("MQTT接收: biz=" + bizId + " topic=" + topic + " msgCount=" + jsonArray.size());
+        System.out.println("MQTT接收: biz=" + bizId + " topic=" + topic + " msgCount=" + jsonArray.size());
+//        MPointHistoryService mPointHistoryService = (MPointHistoryService) SpringContextUtil.getBean("mPointHistoryService");
         for (int i = 0; i < jsonArray.size(); i++) {
             JSONObject jsonObject = jsonArray.getJSONObject(i);
             // 解析时间戳
@@ -135,6 +142,13 @@ public class MqttServiceImpl implements MqttService {
                 // 处理ES数据
                 MPointES esPoint = buildEsPoint(mPoint, value, date);
                 redisList.add(esPoint);
+
+                MPointHistory mPointHistory = new MPointHistory();
+                mPointHistory.setMeasuredt(date);
+                mPointHistory.setParmvalue(value);
+                mPointHistory.setTbName("tb_mp_" + key);
+                mPointHistory.setUserid(port);
+                insertInsert(mPointHistory);
             }
             // 推送前端数据
 //            loggger.info("准备推送前端: topic={} size={} @{}", topic, vueList.size(), date);
@@ -146,6 +160,22 @@ public class MqttServiceImpl implements MqttService {
         }
     }
 
+    @DataSourceTypeAnno(DataSourceEnum.SCADA_JSWS)
+    public void insertInsert(MPointHistory mPointHistory) {
+        try {
+            int exist = mPointHistoryService.checkTableExist(mPointHistory.getTbName());
+            if (exist == 0) { // 不存在就新建点位表
+                loggger.info(String.format("[重要]创建了点位表：%s  %s", mPointHistory.getTbName(), DataSourceContextHolder.getDataSourceType()));
+                mPointHistoryService.createTable(mPointHistory.getTbName());
+            }
+
+            mPointHistoryService.save("", mPointHistory);
+            loggger.info("保存成功一条数据到历史表：{}", mPointHistory);
+        } catch (Exception e) {
+            loggger.error("mPointHistoryService.save error :{}", e.getMessage());
+        }
+    }
+
     /**
      * 校验值是否有效（非空且长度合法）
      */

src/main/java/com/sipai/service/scada/MPointHistoryService.java

@@ -5,7 +5,7 @@ import com.sipai.entity.scada.MPointHistory;
 import java.util.List;
 
 public interface MPointHistoryService {
-    int save(String bizId, MPointHistory mPointHistory);
+    Integer save(String bizId, MPointHistory mPointHistory);
 
     List<MPointHistory> selectListByTableAWhere(String bizId, String table, String wherestr);
 
@@ -14,4 +14,8 @@ public interface MPointHistoryService {
     int saveByCreate(String bizId, MPointHistory mPointHistory);
 
     List<MPointHistory> selectIndustrialLibrary(String bizId, String IP,String MPointID, String intv, String sdt, String edt);
+
+    void createTable(String tbName);
+
+    int checkTableExist(String tbName);
 }

src/main/java/com/sipai/service/scada/impl/MPointBzwServiceImpl.java

@@ -3,6 +3,7 @@ package com.sipai.service.scada.impl;
 import com.sipai.dao.scada.MPointBzwDao;
 import com.sipai.entity.scada.MPointBzw;
 import com.sipai.service.scada.MPointBzwService;
+import com.sipai.tools.DataSourceEnum;
 import com.sipai.tools.DataSourceTypeAnno;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.scheduling.annotation.Async;
@@ -22,32 +23,32 @@ public class MPointBzwServiceImpl implements MPointBzwService {
     private MPointBzwDao mPointBzwDao;
 
     @Override
-    @DataSourceTypeAnno()
+    @DataSourceTypeAnno(DataSourceEnum.SCADA_JSWS)
     public MPointBzw selectByPrimaryKey(String id) {
         return mPointBzwDao.selectByPrimaryKey(id);
     }
 
     @Override
-    @DataSourceTypeAnno()
+    @DataSourceTypeAnno(DataSourceEnum.SCADA_JSWS)
     public Integer deleteByPrimaryKey(String id) {
         return mPointBzwDao.deleteByPrimaryKey(id);
     }
 
     @Override
-    @DataSourceTypeAnno()
+    @DataSourceTypeAnno(DataSourceEnum.SCADA_JSWS)
     public Integer insert(MPointBzw entity) {
         return mPointBzwDao.insert(entity);
     }
 
     @Async
     @Override
-    @DataSourceTypeAnno()
+    @DataSourceTypeAnno(DataSourceEnum.SCADA_JSWS)
     public Integer updateByPrimaryKeySelective(MPointBzw entity) {
         return mPointBzwDao.updateByPrimaryKeySelective(entity);
     }
 
     @Override
-    @DataSourceTypeAnno()
+    @DataSourceTypeAnno(DataSourceEnum.SCADA_JSWS)
     public List<MPointBzw> selectListByWhere(String wherestr) {
         MPointBzw mPoint = new MPointBzw();
         mPoint.setWhere(wherestr);
@@ -55,7 +56,7 @@ public class MPointBzwServiceImpl implements MPointBzwService {
     }
 
     @Override
-    @DataSourceTypeAnno()
+    @DataSourceTypeAnno(DataSourceEnum.SCADA_JSWS)
     public Integer deleteByWhere(String wherestr) {
         return mPointBzwDao.deleteByWhere(wherestr);
     }

src/main/java/com/sipai/service/scada/impl/MPointHistoryServiceImpl.java

@@ -5,10 +5,12 @@ import com.sipai.entity.scada.MPointHistory;
 import com.sipai.service.scada.MPointHistoryService;
 import com.sipai.tools.CommUtil;
 import com.sipai.tools.DataSourceContextHolder;
+import com.sipai.tools.DataSourceEnum;
 import com.sipai.tools.DataSourceTypeAnno;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Service;
+import org.springframework.util.StringUtils;
 
 import java.util.List;
 
@@ -19,11 +21,11 @@ public class MPointHistoryServiceImpl implements MPointHistoryService {
     @Autowired
     MPointHistoryDao mPointHistoryDao;
 
-    @DataSourceTypeAnno()
+    @DataSourceTypeAnno(DataSourceEnum.SCADA_JSWS)
     @Override
-    public int save(String bizId, MPointHistory mPointHistory) {
+    public Integer save(String bizId, MPointHistory mPointHistory) {
         try {
-            int res = mPointHistoryDao.insert(mPointHistory);
+            Integer res = mPointHistoryDao.insert(mPointHistory);
             return res;
         } catch (Exception e) {
             System.out.println(e);
@@ -31,38 +33,43 @@ public class MPointHistoryServiceImpl implements MPointHistoryService {
         }
     }
 
-    @DataSourceTypeAnno()
+    @DataSourceTypeAnno(DataSourceEnum.SCADA_JSWS)
     @Override
     public List<MPointHistory> selectListByTableAWhere(String bizId, String table, String wherestr) {
-        return mPointHistoryDao.selectListByTableAWhere(table ,wherestr);
+        return mPointHistoryDao.selectListByTableAWhere(table, wherestr);
     }
 
-    @DataSourceTypeAnno()
+    @DataSourceTypeAnno(DataSourceEnum.SCADA_JSWS)
     @Override
-    public int deleteByTableAWhere(String bizId,String table ,String wherestr) {
+    public int deleteByTableAWhere(String bizId, String table, String wherestr) {
-        return mPointHistoryDao.deleteByTableAWhere(table ,wherestr);
+        return mPointHistoryDao.deleteByTableAWhere(table, wherestr);
     }
 
-    @DataSourceTypeAnno()
+    @DataSourceTypeAnno(DataSourceEnum.SCADA_JSWS)
     @Override
     public int saveByCreate(String bizId, MPointHistory mPointHistory) {
         mPointHistory.setTbName(CommUtil.getMPointTableName(mPointHistory.getTbName()));
-        if(CommUtil.isBlank(bizId)){
+        if (CommUtil.isBlank(bizId)) {
             return 0;
         }
         try {
             int i = this.mPointHistoryDao.checkTableExist(mPointHistory.getTbName());
-            if(i == 0) {
+            if (i == 0) {
-                log.info(String.format("[重要]创建了表：%s  %s",mPointHistory.getTbName(), DataSourceContextHolder.getDataSourceType()));
+                log.info(String.format("[重要]创建了表：%s  %s", mPointHistory.getTbName(), DataSourceContextHolder.getDataSourceType()));
                 mPointHistoryDao.createTable(mPointHistory.getTbName());
             }
 
             // 添加时间判断 todo 将来优化
-            mPointHistory.setWhere(" where MeasureDT = '" + mPointHistory.getMeasuredt() + "'");
+            if (StringUtils.isEmpty(mPointHistory.getMeasuredt())) {
+                mPointHistory.setWhere(" where MeasureDT IS NULL");
+            } else {
+                mPointHistory.setWhere(" where MeasureDT = '" + mPointHistory.getMeasuredt() + "'");
+            }
             Integer count = this.mPointHistoryDao.selectCount(mPointHistory);
 
-            int result = count == 0 ? mPointHistoryDao.insert(mPointHistory) : mPointHistoryDao.updateByMeasureDt(mPointHistory);
+//            int result = count == 0 ? mPointHistoryDao.insert(mPointHistory) : mPointHistoryDao.updateByMeasureDt(mPointHistory);
-            return result;
+//            return result;
+            return 0;
         } catch (Exception e) {
             System.out.println(DataSourceContextHolder.getDataSourceType() + "   |   " + e);
             // 返回默认值
@@ -70,10 +77,21 @@ public class MPointHistoryServiceImpl implements MPointHistoryService {
         }
     }
 
-    @DataSourceTypeAnno()
+    @DataSourceTypeAnno(DataSourceEnum.SCADA_JSWS)
     @Override
     public List<MPointHistory> selectIndustrialLibrary(String bizId, String IP, String MPointID, String intv, String sdt, String edt) {
         return mPointHistoryDao.selectIndustrialLibrary(IP, MPointID, intv, sdt, edt);
     }
 
+    @DataSourceTypeAnno(DataSourceEnum.SCADA_JSWS)
+    @Override
+    public void createTable(String tbName) {
+        mPointHistoryDao.createTable(tbName);
+    }
+
+    @DataSourceTypeAnno(DataSourceEnum.SCADA_JSWS)
+    @Override
+    public int checkTableExist(String tbName) {
+        return mPointHistoryDao.checkTableExist(tbName);
+    }
 }

src/main/java/com/sipai/service/scada/impl/MPointServiceImpl.java

@@ -13,6 +13,7 @@ import com.sipai.service.scada.MPointHistoryService;
 import com.sipai.service.scada.MPointService;
 import com.sipai.tools.CommString;
 import com.sipai.tools.CommUtil;
+import com.sipai.tools.DataSourceEnum;
 import com.sipai.tools.DataSourceTypeAnno;
 import lombok.extern.slf4j.Slf4j;
 import org.elasticsearch.index.query.BoolQueryBuilder;
@@ -81,7 +82,7 @@ public class MPointServiceImpl implements MPointService {
         return mPointDao.selectByPrimaryKey(id);
     }*/
 
-    @DataSourceTypeAnno()
+    @DataSourceTypeAnno(DataSourceEnum.SCADA_JSWS)
     public MPoint selectById(String unitId, String id) {
         return mPointDao.selectByPrimaryKey(id);
     }
@@ -92,6 +93,7 @@ public class MPointServiceImpl implements MPointService {
      * @param id
      * @return
      */
+    @DataSourceTypeAnno(DataSourceEnum.SCADA_JSWS)
     @Override
     public MPoint selectById4Es(String id) {
         if (id == null || id.isEmpty()) {
@@ -122,7 +124,7 @@ public class MPointServiceImpl implements MPointService {
 
     }
 
-    //    @DataSourceTypeAnno()
+    @DataSourceTypeAnno(DataSourceEnum.SCADA_JSWS)
     @Override
     public List<MPoint> selectListByWhere(String wherestr) {
         MPoint mpoint = new MPoint();
@@ -145,6 +147,7 @@ public class MPointServiceImpl implements MPointService {
     }
 
     @Override
+    @DataSourceTypeAnno(DataSourceEnum.SCADA_JSWS)
     public List<MPoint> selectListByES(Map<String, Object> map, Integer page, Integer rows) {
         List<MPoint> mpoints = new ArrayList<>();
         try {
@@ -187,7 +190,7 @@ public class MPointServiceImpl implements MPointService {
         return mpoints;
     }
 
-    //    @DataSourceTypeAnno()
+    @DataSourceTypeAnno(DataSourceEnum.SCADA_JSWS)
     @Override
     public int update(MPoint entity) {
         try {
@@ -199,7 +202,7 @@ public class MPointServiceImpl implements MPointService {
         }
     }
 
-    @DataSourceTypeAnno()
+    @DataSourceTypeAnno(DataSourceEnum.SCADA_JSWS)
     @Override
     public int updateValueByKey(String bizId, MPoint entity) {
         try {
@@ -248,6 +251,7 @@ public class MPointServiceImpl implements MPointService {
      * @param entity
      * @return
      */
+    @DataSourceTypeAnno(DataSourceEnum.SCADA_JSWS)
     public int saveInfluxdb(String bizId, MPointHistory entity) {
         Map<String, Object> fields = new HashMap<>();
         Map<String, String> tags = new HashMap<>();
@@ -261,10 +265,10 @@ public class MPointServiceImpl implements MPointService {
         long time = Calendar.getInstance().getTimeInMillis();
         try {
             fields.put("parmvalue", Float.parseFloat(entity.getParmvalue().toString()));
-            time = longDateFormat.parse(entity.getMeasuredt()).getTime();
+//            time = longDateFormat.parse(entity.getMeasuredt()).getTime();
 //            influxDBConfig.insert("tb_mp_" + entity.getTbName(), time, tags, fields);
 //            System.out.println("===插入数据成功===" + entity.getTbName() + "===" + Float.parseFloat(entity.getParmvalue().toString()));
-        } catch (ParseException e) {
+        } catch (Exception e) {
             e.printStackTrace();
         }
         return 1;
@@ -276,6 +280,7 @@ public class MPointServiceImpl implements MPointService {
      * @param mPointES
      */
 //    @Async
+    @DataSourceTypeAnno(DataSourceEnum.SCADA_JSWS)
     @Override
     public void saveAsync(List<MPointES> mPointES) {
         this.mPointRepo.batchUpdate(mPointES);
@@ -286,6 +291,7 @@ public class MPointServiceImpl implements MPointService {
      *
      * @return
      */
+    @DataSourceTypeAnno(DataSourceEnum.SCADA_JSWS)
     public List<MPoint> selectListByWhere4Es(NativeSearchQueryBuilder nativeSearchQueryBuilder) {
         SearchQuery searchQuery = nativeSearchQueryBuilder.build();
         Page<MPoint> mPage = mPointRepo.search(searchQuery);
@@ -312,6 +318,7 @@ public class MPointServiceImpl implements MPointService {
      * @return
      * @throws Exception
      */
+    @DataSourceTypeAnno(DataSourceEnum.SCADA_JSWS)
     public static String formatTimeEight(String time) throws Exception {
         Date d = null;
         SimpleDateFormat sd = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
@@ -321,6 +328,7 @@ public class MPointServiceImpl implements MPointService {
         return newtime;
     }
 
+    @DataSourceTypeAnno(DataSourceEnum.SCADA_JSWS)
     public void sendKafka4OpcUa(String id, Double value, String time) {
         JSONObject message = new JSONObject();
         message.put("id", id);
@@ -365,6 +373,7 @@ public class MPointServiceImpl implements MPointService {
         }
     }
 
+    @DataSourceTypeAnno(DataSourceEnum.SCADA_JSWS)
     @Override
     public void saveRedis(String mpcode, Object value, String time) {
         RBatch batch = redissonClient.createBatch();
@@ -376,7 +385,8 @@ public class MPointServiceImpl implements MPointService {
         batch.execute();
     }
 
-    public void sendKafka4UDP(String unitId,String mpcode, Object value, String timePoint, int i, int length) {
+    @DataSourceTypeAnno(DataSourceEnum.SCADA_JSWS)
+    public void sendKafka4UDP(String unitId, String mpcode, Object value, String timePoint, int i, int length) {
 
         JSONObject message = new JSONObject();
         message.put("id", mpcode);
@@ -408,7 +418,7 @@ public class MPointServiceImpl implements MPointService {
         try {
             MPointHistory mPointHistory = new MPointHistory();
             mPointHistory.setParmvalue(CommUtil.toBigDecimal(value));
-            mPointHistory.setMeasuredt(timePoint);
+//            mPointHistory.setMeasuredt(timePoint);
             mPointHistory.setTbName(mpcode);
             mPointHistory.setUserid("datacollector");
             mPointHistory.setInsdt(CommUtil.nowDate());
@@ -421,7 +431,7 @@ public class MPointServiceImpl implements MPointService {
         //更新最后一条数据到总表
         if (i == length - 1) {
             try {
-                MPoint mPoint = mPointService.selectById(unitId,mpcode);
+                MPoint mPoint = mPointService.selectById(unitId, mpcode);
                 if (mPoint != null) {
                     mPoint.setParmvalue(CommUtil.toBigDecimal(value));
                     mPoint.setMeasuredt(timePoint);
@@ -437,14 +447,15 @@ public class MPointServiceImpl implements MPointService {
     }
 
     //    @Async
+    @DataSourceTypeAnno(DataSourceEnum.SCADA_JSWS)
     @Override
-    public void sendKafka4MQTT(String unitId,String mpcode, Object value, String timePoint, int i, int length, String signaltype) {
+    public void sendKafka4MQTT(String unitId, String mpcode, Object value, String timePoint, int i, int length, String signaltype) {
         //写子表
         if (signaltype != null && signaltype.equals("DI")) {
             try {
                 MPointHistory mPointHistory = new MPointHistory();
                 mPointHistory.setParmvalue(CommUtil.toBigDecimal(value));
-                mPointHistory.setMeasuredt(timePoint);
+//                mPointHistory.setMeasuredt(timePoint);
                 mPointHistory.setTbName(mpcode);
                 mPointHistory.setUserid("datacollector");
                 mPointHistory.setInsdt(CommUtil.nowDate());
@@ -457,7 +468,7 @@ public class MPointServiceImpl implements MPointService {
         //更新最后一条数据到总表
         if (i == length - 1) {
             try {
-                MPoint mPoint = mPointService.selectById(unitId,mpcode);
+                MPoint mPoint = mPointService.selectById(unitId, mpcode);
                 if (mPoint != null) {
                     mPoint.setParmvalue(CommUtil.toBigDecimal(value));
                     mPoint.setMeasuredt(timePoint);
@@ -470,6 +481,7 @@ public class MPointServiceImpl implements MPointService {
     }
 
     //    @Async
+    @DataSourceTypeAnno(DataSourceEnum.SCADA_JSWS)
     @Override
     public void sendKafka4MQTT_Alarm(String mpcode, Object value, String timePoint, int i, int length) {
         JSONObject message = new JSONObject();

src/main/java/com/sipai/tools/CommUtil.java

@@ -181,7 +181,7 @@ public class CommUtil {
             format = "yyyy-MM-dd HH:mm:ss";
         }
         SimpleDateFormat sdf = new SimpleDateFormat(format);
-        return sdf.format(new Date(Long.valueOf(seconds + "000")));
+        return sdf.format(new Date(Long.valueOf(seconds)));
     }
 
     /**

src/main/java/com/sipai/tools/DataSourceAspect.java

@@ -29,23 +29,23 @@ public class DataSourceAspect {
         MethodSignature methodSignature = (MethodSignature) pjp.getSignature();
         Method method = methodSignature.getMethod();
         DataSourceTypeAnno typeAnno = method.getAnnotation(DataSourceTypeAnno.class);
-        DataSourceEnum sourceEnum = typeAnno.value();
+        DataSourceEnum sourceEnum = typeAnno.value(); 
-
         Object[] argusObjects = pjp.getArgs();
         Object argus = argusObjects[0];
-
+        String packageName = method.getDeclaringClass().getName();
-//        if (argus == DataSourceEnum.master) {
+        if (sourceEnum == DataSourceEnum.SCADA_JSWS || packageName.contains(".scada.")) {
-//            DataSourceContextHolder.setDataSourceType(DataSourceEnum.master);
+            DataSourceContextHolder.setDataSourceType(DataSourceEnum.SCADA_JSWS);
-//        } else if (sourceEnum == DataSourceEnum.slaver) {
+        } else  {
-//            DataSourceContextHolder.setDataSourceType(DataSourceEnum.slaver);
-//        }
-
-        if (argus != null && !argus.equals("") && argusObjects.length >= 2) {
-            //DataSourceContextHolder.setDataSourceType(DataSourceEnum.valueOf("SCADA_" + argus.toString()));
-            DataSourceContextHolder.setDataSourceType(DataSourceEnum.valueOf("SCADA_" + argus.toString()));
-        } else {
             DataSourceContextHolder.setDataSourceType(DataSourceEnum.master);
         }
+
+//        if (argus != null && !argus.equals("") && argusObjects.length >= 2) {
+//            //DataSourceContextHolder.setDataSourceType(DataSourceEnum.valueOf("SCADA_" + argus.toString()));
+//            DataSourceContextHolder.setDataSourceType(DataSourceEnum.valueOf("SCADA_" + argus.toString()));
+//        } else {
+//            DataSourceContextHolder.setDataSourceType(DataSourceEnum.master);
+//        }
+
         Object result = null;
         try {
             result = pjp.proceed();

src/main/java/com/sipai/tools/DataSourceEnum.java

@@ -6,5 +6,6 @@ package com.sipai.tools;
 public enum DataSourceEnum {
     master
     //
-    ,SCADA_JS1C,SCADA_JS2C,SCADA_JSBZ
+//    ,SCADA_JS1C,SCADA_JS2C,SCADA_JSBZ,
+    ,SCADA_JSWS,
 }

src/main/java/com/sipai/tools/DataSourceTypeAnno.java

@@ -13,4 +13,5 @@ import java.lang.annotation.Target;
 public @interface DataSourceTypeAnno {
     //使用方式在service层方法上添加@DataSourceTypeAnno(DataSourceEnum.数据源枚举类型)用于指定所使用的数据源
     DataSourceEnum value() default DataSourceEnum.master;
+//    DataSourceEnum value() default DataSourceEnum.SCADA_JSWS;
 }

src/main/java/com/sipai/tools/MyBatisConfig.java

@@ -35,28 +35,35 @@ public class MyBatisConfig {
     /**
      * 下面为管网
      */
-    @Bean("SCADA_JS1C")
+//    @Bean("SCADA_JS1C")
-    @ConfigurationProperties(prefix = "spring.datasource.js1")
+//    @ConfigurationProperties(prefix = "spring.datasource.js1")
     //根据数据源前缀到application.yml读取数据源信息//可以配置更多数据源，到前提是application.yml中存在，而且也需要在枚举类中添加枚举类型
-    public DataSource fssk() throws Exception {
+//    public DataSource fssk() throws Exception {
-        return DataSourceBuilder.create().build();
+//        return DataSourceBuilder.create().build();
-    }
+//    }
 
     /**
      * 下面为管网
      */
-    @Bean("SCADA_JS2C")
+//    @Bean("SCADA_JS2C")
-    @ConfigurationProperties(prefix = "spring.datasource.js2")
+//    @ConfigurationProperties(prefix = "spring.datasource.js2")
     //根据数据源前缀到application.yml读取数据源信息//可以配置更多数据源，到前提是application.yml中存在，而且也需要在枚举类中添加枚举类型
-    public DataSource fssk2() throws Exception {
+//    public DataSource fssk2() throws Exception {
-        return DataSourceBuilder.create().build();
+//        return DataSourceBuilder.create().build();
-    }
+//    }
 
     /**
      * 下面为管网
      */
-    @Bean("SCADA_JSBZ")
+//    @Bean("SCADA_JSBZ")
-    @ConfigurationProperties(prefix = "spring.datasource.js3")
+//    @ConfigurationProperties(prefix = "spring.datasource.js3")
+    //根据数据源前缀到application.yml读取数据源信息//可以配置更多数据源，到前提是application.yml中存在，而且也需要在枚举类中添加枚举类型
+//    public DataSource fssk3() throws Exception {
+//        return DataSourceBuilder.create().build();
+//    }
+
+    @Bean("SCADA_JSWS")
+    @ConfigurationProperties(prefix = "spring.datasource.jsws")
     //根据数据源前缀到application.yml读取数据源信息//可以配置更多数据源，到前提是application.yml中存在，而且也需要在枚举类中添加枚举类型
     public DataSource fssk3() throws Exception {
         return DataSourceBuilder.create().build();
@@ -68,15 +75,18 @@ public class MyBatisConfig {
      */
     @Bean("dynamicDataSource")
     public DynamicDataSource dynamicDataSource(@Qualifier("masterDataSource") DataSource masterDataSource,
-                                               @Qualifier("SCADA_JS1C") DataSource SCADA_JS1C,
+//                                               @Qualifier("SCADA_JS1C") DataSource SCADA_JS1C,
-                                               @Qualifier("SCADA_JS2C") DataSource SCADA_JS2C,
+//                                               @Qualifier("SCADA_JS2C") DataSource SCADA_JS2C,
-                                               @Qualifier("SCADA_JSBZ") DataSource SCADA_JSBZ) {
+//                                               @Qualifier("SCADA_JSBZ") DataSource SCADA_JSBZ
+                                               @Qualifier("SCADA_JSWS") DataSource SCADA_JSWS
+    ) {
         Map<Object, Object> targetDataSources = new HashMap<Object, Object>();
         targetDataSources.put(DataSourceEnum.master, masterDataSource);
 
-        targetDataSources.put(DataSourceEnum.SCADA_JS1C, SCADA_JS1C);
+//        targetDataSources.put(DataSourceEnum.SCADA_JS1C, SCADA_JS1C);
-        targetDataSources.put(DataSourceEnum.SCADA_JS2C, SCADA_JS2C);
+//        targetDataSources.put(DataSourceEnum.SCADA_JS2C, SCADA_JS2C);
-        targetDataSources.put(DataSourceEnum.SCADA_JSBZ, SCADA_JSBZ);
+//        targetDataSources.put(DataSourceEnum.SCADA_JSBZ, SCADA_JSBZ);
+        targetDataSources.put(DataSourceEnum.SCADA_JSWS, SCADA_JSWS);
 
         DynamicDataSource dataSource = new DynamicDataSource();
         dataSource.setTargetDataSources(targetDataSources);// 该方法是AbstractRoutingDataSource的方法

src/main/resources/application-dev.yml

@@ -13,49 +13,57 @@ server:
 spring:
   redis:
     mode: single  #单服务模式
-    host: 132.120.136.19
+    host: 122.51.194.184
-    port: 6379
+    port: 26739
     database: 0
     cluster:
       nodes: 172.16.242.44:6379,172.16.242.45:6379,172.16.242.46:6379,172.16.242.47:6379,172.16.242.48:6379,172.16.242.49:6379
       max-redirects: 3
-    password: zaqwsx
+    password: Aa112211
     timeout: 100000
 
   data:
     elasticsearch:
-      cluster-name: elasticsearch-sipaiis
+#      cluster-name: elasticsearch-sipaiis
-      cluster-nodes: 132.120.136.19:9200
+      cluster-name: my-application
+      cluster-nodes: 127.0.0.1:9300
 
   datasource:
     master:
       username: sa
-      password: ZAQwsx@2016
+      password: P76XB3nm36aMkN6n
-      jdbc-url: jdbc:sqlserver://132.120.136.19:1433;DatabaseName=JSYW
+      jdbc-url: jdbc:sqlserver://122.51.194.184:1433;DatabaseName=SIPAIIS_WMS_HQAQ
       driver-class-name: com.microsoft.sqlserver.jdbc.SQLServerDriver
       type: com.alibaba.druid.pool.DruidDataSource
 
     #生产库
-    js1:
+#    js1:
-      username: sa
+#      username: sa
-      password: ZAQwsx@2016
+#      password: P76XB3nm36aMkN6n
-      jdbc-url: jdbc:sqlserver://132.120.136.19:1433;DatabaseName=EIP_PRD_JS1C
+#      jdbc-url: jdbc:sqlserver://122.51.194.184:1433;DatabaseName=EIP_PRD_HQWS
-      driver-class-name: com.microsoft.sqlserver.jdbc.SQLServerDriver
+#      driver-class-name: com.microsoft.sqlserver.jdbc.SQLServerDriver
-      type: com.alibaba.druid.pool.DruidDataSource
+#      type: com.alibaba.druid.pool.DruidDataSource
 
     #生产库
-    js2:
+#    js2:
-      username: sa
+#      username: sa
-      password: ZAQwsx@2016
+#      password: P76XB3nm36aMkN6n
-      jdbc-url: jdbc:sqlserver://132.120.136.19:1433;DatabaseName=EIP_PRD_JS1C
+#      jdbc-url: jdbc:sqlserver://122.51.194.184:1433;DatabaseName=EIP_PRD_HQWS
-      driver-class-name: com.microsoft.sqlserver.jdbc.SQLServerDriver
+#      driver-class-name: com.microsoft.sqlserver.jdbc.SQLServerDriver
-      type: com.alibaba.druid.pool.DruidDataSource
+#      type: com.alibaba.druid.pool.DruidDataSource
 
     #生产库
-    js3:
+#    js3:
+#      username: sa
+#      password: P76XB3nm36aMkN6n
+#      jdbc-url: jdbc:sqlserver://122.51.194.184:1433;DatabaseName=EIP_PRD_HQWS
+#      driver-class-name: com.microsoft.sqlserver.jdbc.SQLServerDriver
+#      type: com.alibaba.druid.pool.DruidDataSource
+
+    jsws:
       username: sa
-      password: ZAQwsx@2016
+      password: P76XB3nm36aMkN6n
-      jdbc-url: jdbc:sqlserver://132.120.136.19:1433;DatabaseName=EIP_PRD_JS1C
+      jdbc-url: jdbc:sqlserver://122.51.194.184:1433;DatabaseName=EIP_PRD_HQWS
       driver-class-name: com.microsoft.sqlserver.jdbc.SQLServerDriver
       type: com.alibaba.druid.pool.DruidDataSource
 
@@ -145,10 +153,14 @@ app:
   max-points: 200 #实时数据接口一次最多点位数
 
 mqtt:
-  broker-address: "tcp://192.168.10.17:1883"
+  broker-address: "tcp://122.51.194.184:1883"
-  dashboard-01: "http://192.168.10.17:18083"
+  dashboard-01: "http://122.51.194.184:18083"
-  username: "admin"
+  username: "dmbroker"
-  password: "sipai@64368180"
+  password: "qwer1234"
+#  broker-address: "tcp://121.5.164.6:1883"
+#  dashboard-01: "http://121.5.164.6:18083"
+#  username: "dmbroker"
+#  password: "qwer1234"
 
 # udp服务
 udp:

src/main/resources/application.yml

@@ -1,4 +1,4 @@
 spring:
   profiles:
-#    active: dev
+    active: dev
-    active: pro
+#    active: pro

src/main/resources/mybatis/mapper/scada/MPointHistoryMapper.xml

@@ -12,7 +12,7 @@
     </resultMap>
 
     <insert id="insert" parameterType="com.sipai.entity.scada.MPointHistory">
-        insert into `${tbName}` (ParmValue, MeasureDT,
+        insert into ${tbName} (ParmValue, MeasureDT,
                                  memotype, memo, userid,
                                  insdt)
         values (#{parmvalue,jdbcType=DECIMAL}, #{measuredt,jdbcType=TIMESTAMP},
@@ -39,27 +39,26 @@
 
     <select id="selectCount" parameterType="com.sipai.entity.scada.MPointHistory" resultType="java.lang.Integer">
         select count(*)
-        from `${tbName}`
+        from ${tbName}
                  ${where}
     </select>
 
     <update id="createTable" parameterType="java.lang.String">
-        CREATE TABLE `${table}`
+        CREATE TABLE ${table} (
-        (
+            ItemID bigint IDENTITY(1,1) NOT NULL,
-            `ItemID`    bigint          NOT NULL AUTO_INCREMENT,
+            ParmValue DECIMAL(18, 4) NOT NULL,
-            `ParmValue` decimal(18, 4) NOT NULL,
+            MeasureDT datetime NOT NULL,
-            `MeasureDT` datetime        NOT NULL,
+            memotype VARCHAR(100) NULL,
-            `memotype`  varchar(100) NULL,
+            memo VARCHAR(50) NULL,
-            `memo`      varchar(50) NULL,
+            userid VARCHAR(50) NULL,
-            `userid`    varchar(50) NULL,
+            insdt datetime NULL,
-            `insdt`     datetime NULL ON UPDATE CURRENT_TIMESTAMP,
+            CONSTRAINT PK_${table} PRIMARY KEY (ItemID)
-            PRIMARY KEY (`ItemID`),
-            INDEX `index_dt`(`MeasureDT`)
         );
+        CREATE INDEX index_dt ON ${table} (MeasureDT);
     </update>
 
     <update id="updateByMeasureDt" parameterType="com.sipai.entity.scada.MPointHistory">
-        update `${tbName}`
+        update ${tbName}
         set ParmValue = #{parmvalue,jdbcType=DECIMAL}, memotype = #{memotype,jdbcType=VARCHAR}
         where MeasureDT = #{measuredt,jdbcType=TIMESTAMP}
     </update>

src/main/resources/mybatis/mapper/scada/MPointMapper.xml

@@ -682,7 +682,7 @@
   <select id="selectListByWhere" parameterType="java.lang.String" resultMap="BaseResultMap">
 		select 
 		<include refid="Base_Column_List" />
-		from tb_measurepoint
+		from TB_MeasurePoint
 		${where}
 	</select>
 	<delete id="deleteByWhere" parameterType="java.lang.String">