Merge remote-tracking branch 'origin/master'

wcs/nladmin-system/src/main/java/org/nl/acs/opc/DeviceOpcProtocolRunable.java

@@ -1,29 +1,38 @@
 package org.nl.acs.opc;
 
 import cn.hutool.core.date.DateUtil;
-import cn.hutool.core.util.ObjectUtil;
 import cn.hutool.core.util.StrUtil;
 import lombok.extern.slf4j.Slf4j;
 import org.nl.acs.udw.UnifiedDataAccessor;
 import org.nl.acs.udw.UnifiedDataAccessorFactory;
-import org.openscada.opc.lib.da.Group;
+import org.nl.acs.udw.UnifiedDataAppService;
-import org.openscada.opc.lib.da.Item;
+import org.nl.modules.lucene.service.LuceneExecuteLogService;
-import org.openscada.opc.lib.da.ItemState;
+import org.nl.modules.lucene.service.dto.LuceneLogDto;
-import org.openscada.opc.lib.da.Server;
+import org.nl.modules.wql.util.SpringContextHolder;
+import org.openscada.opc.lib.da.*;
+import org.springframework.beans.factory.annotation.Autowired;
 
 import java.util.*;
 
+
 @Slf4j
-public class DeviceOpcProtocolRunable implements Runnable {
+public class DeviceOpcProtocolRunable implements Runnable, DataCallback, ServerConnectionStateListener {
   List<OpcItemDto> protocols;
   OpcServerManageDto OpcServer;
   int error_num;
   String message;
+  int maxResartNum;
   private Server server;
+  private Group group;
+  boolean flag = false;
+  private int all_null;
+  private Map<String, OpcItemDto> itemSearchCache;
 
   public DeviceOpcProtocolRunable() {
     this.error_num = 0;
+    this.all_null = 0;
     this.message = null;
+    this.itemSearchCache = new HashMap();
     this.server = null;
   }
 
@@ -43,36 +52,58 @@ public class DeviceOpcProtocolRunable implements Runnable {
     this.OpcServer = opcServer;
   }
 
-  OpcItemDto getItem(String item) {
-    Iterator var2 = this.protocols.iterator();
 
-    OpcItemDto dto;
+  private OpcItemDto getItem(String item) {
-    do {
+    OpcItemDto x = (OpcItemDto) this.itemSearchCache.get(item);
-      if (!var2.hasNext()) {
+    if (x == null) {
-        return null;
+      Iterator var3 = this.protocols.iterator();
+
+      while (var3.hasNext()) {
+        OpcItemDto dto = (OpcItemDto) var3.next();
+        if (StrUtil.equals(item, dto.getItem_code())) {
+          x = dto;
+          this.itemSearchCache.put(item, dto);
+          break;
+        }
       }
+    }
 
-      dto = (OpcItemDto) var2.next();
+    return x;
-    } while (!StrUtil.equals(item, dto.getItem_code()));
-
-    return dto;
   }
 
+
   @Override
   public void run() {
-    while (true) {
+    if (OpcConfig.opc_item_read_using_callback) {
-      try {
+      this.runNew();
-         this.server = OpcServerUtl.getServerWithOutException(this.OpcServer.getOpc_host(), this.OpcServer.getCls_id(), this.OpcServer.getUser(), this.OpcServer.getPassword(), this.OpcServer.getDomain());
+    } else {
-//        Group group = server.addGroup(this.OpcServer.getOpc_host());
+      this.runOld();
-        Group group = server.addGroup(this.OpcServer.getOpc_code());
+    }
-        if(ObjectUtil.isEmpty(group)){
+  }
-          log.info("group is null ");
-        }
-        List<String> itemsString = new ArrayList();
-        Iterator it = this.protocols.iterator();
 
-        while (it.hasNext()) {
+
-          OpcItemDto protocol = (OpcItemDto) it.next();
+  private void runOld() {
+    while (true) {
+      start:
+      try {
+        if (this.group != null) {
+          group.clear();
+          group.remove();
+          log.trace("清理group...");
+        }
+        if (this.server != null) {
+          server.disconnect();
+          log.trace("清理server...");
+        }
+
+        this.server = OpcServerUtl.getServerWithOutException(this.OpcServer.getOpc_host(), this.OpcServer.getCls_id(), this.OpcServer.getUser(), this.OpcServer.getPassword(), this.OpcServer.getDomain());
+        this.server.addStateListener(this);
+        group = this.server.addGroup();
+        List<String> itemsString = new ArrayList();
+        Iterator var3 = this.protocols.iterator();
+
+        while (var3.hasNext()) {
+          OpcItemDto protocol = (OpcItemDto) var3.next();
           String item = protocol.getItem_code();
           itemsString.add(item);
         }
@@ -80,58 +111,60 @@ public class DeviceOpcProtocolRunable implements Runnable {
         Map<String, Item> itemsMap = new LinkedHashMap();
         boolean is_error = false;
         StringBuilder err_message = new StringBuilder();
-        Iterator var7 = itemsString.iterator();
+        Iterator var6 = itemsString.iterator();
 
-        while (var7.hasNext()) {
+        while (var6.hasNext()) {
-          String string = (String) var7.next();
+          String string = (String) var6.next();
 
           try {
-            itemsMap.put(string, group.addItem(string));
+            Item item = group.addItem(string);
+            itemsMap.put(string, item);
             log.trace("添加成功 {}", string);
-          } catch (Exception var29) {
+          } catch (Exception var26) {
-            err_message.append(string + ":" + var29.getMessage());
+            err_message.append(string + ":" + var26.getMessage());
             if (!is_error) {
               is_error = true;
             }
           }
         }
 
+        String tag;
         if (is_error) {
-          log.info("设备OPC数据同步配置异常");
+          tag = err_message.toString();
+          log.warn("{}:{}", OpcConfig.resource_code, tag);
         }
 
         if (!OpcStartTag.is_run) {
           OpcStartTag.is_run = true;
         }
 
-        //线程名
+        tag = "";
-        String tag = Thread.currentThread().getName();
+        if (log.isWarnEnabled()) {
-        if (this.OpcServer != null) {
+          tag = Thread.currentThread().getName();
-          tag = tag + this.OpcServer.getOpc_code();
+          if (this.OpcServer != null) {
+            tag = tag + this.getOpcGroupID();
+          }
         }
-        UnifiedDataAccessor accessor_value = UnifiedDataAccessorFactory.getAccessor(OpcConfig.udw_opc_value_key);
 
+        UnifiedDataAccessor accessor_value = UnifiedDataAccessorFactory.getAccessor(OpcConfig.udw_opc_value_key);
         boolean time_out = false;
 
-        label97:
+        while (DeviceOpcSynchronizeAutoRun.isRun) {
-        while (true) {
           long begin = System.currentTimeMillis();
-          Map<Item, ItemState> itemStatus = null;
+          if (log.isTraceEnabled()) {
-          try {
+            log.trace("{} 开始记时{}", tag, DateUtil.now());
-            itemStatus  = group.read(true, (Item[]) itemsMap.values().toArray(new Item[0]));
-          } catch (Exception e){
-            System.out.println("数据同步异常:"+ this.getOpcServer().getOpc_code());
-            log.trace("数据同步异常:{}", this.getOpcServer().getOpc_code());
-            //e.printStackTrace();
           }
+
+          Map<Item, ItemState> itemStatus = group.read(true, (Item[]) itemsMap.values().toArray(new Item[0]));
           long end = System.currentTimeMillis();
-          log.trace("{} 开始记时{}", tag, DateUtil.now());
           long duration = end - begin;
-          log.trace("{} 读取耗时：{}", tag, duration);
+          if (log.isTraceEnabled()) {
-//          System.out.println("线程："+tag + " 读取耗时:"+ duration);
+            log.trace("{} 读取耗时：{}", tag, duration);
+          }
+
           if (duration > 1000L) {
             if (!time_out) {
-              log.warn(" {} 读取超时 : {}", tag, duration);
+              log.warn("{} 读取超时 : {}", tag, duration);
             }
 
             time_out = true;
@@ -139,99 +172,195 @@ public class DeviceOpcProtocolRunable implements Runnable {
             time_out = false;
           }
 
-          if(ObjectUtil.isEmpty(itemStatus)) {
+          boolean valueAllNotNull = false;
-            //System.out.println( tag + " :itemStatus is null");
-            //log.warn(" {} 读取异常 : {} itemStatus is null", tag);
-          }
-
-
           Set<Item> items = itemStatus.keySet();
           Iterator var18 = items.iterator();
 
-          while (true) {
+          while (var18.hasNext()) {
-            Item item;
+            Item item = (Item) var18.next();
-            //当前值
+            ItemState itemState = (ItemState) itemStatus.get(item);
-            Object value;
+            Object value = OpcUtl.getValue(item, itemState);
-            //旧的值
+            if (value != null) {
-            Object his;
+              valueAllNotNull = true;
-            do {
-              if (!var18.hasNext()) {
-                end = System.currentTimeMillis();
-                log.trace("{}", itemsString);
-                log.trace("{} 计算完成耗时{}", tag, end - begin);
-                Thread.sleep((long) OpcConfig.synchronized_millisecond);
-                if (this.error_num != 0) {
-                  this.error_num = 0;
-                  this.message = null;
-                }
-                continue label97;
-              }
-
-              item = (Item) var18.next();
-              ItemState itemState = (ItemState) itemStatus.get(item);
-              value = OpcUtl.getValue(item, itemState);
-              his = accessor_value.getValue(item.getId());
-              if (!ObjectUtl.isEquals(itemState.getQuality(), QualityTypeValue.OPC_QUALITY_GOOD) && his != null) {
-                log.warn("opc 值不健康 item: {}, 状态: {},当前读取值：{}, 系统内存值{} ", item.getId(), itemState.getQuality(), value, his);
-              }
-            } while (ObjectUtil.equal(value, his));//如果两次的值相等，不走下面的代码
-
-            OpcItemDto itemDto = this.getItem(item.getId());
-            //默认记录日志
-            if (true) {
-              StringBuilder sb = new StringBuilder();
-              //设备的ITEM项
-              List<String> relate_items = itemDto.getRelate_items();
-              Iterator var26 = relate_items.iterator();
-
-              while (var26.hasNext()) {
-                String relate = (String) var26.next();
-                Object obj = accessor_value.getValue(relate);
-                sb.append("key:" + relate + "value:" + obj + ";");
-              }
-
-              log.warn("信号{}变更从{}->{};信号快照：{}", new Object[]{item.getId(), his, value, sb});
-
             }
-//            accessor_value.setValueWithPersistence(item.getId(),accessor_value.getValue(item.getId()));
-//            accessor_value.getHistoryUnifiedData(item.getId());
-            //设置值
-            accessor_value.setValue(item.getId(), value);
 
+            String itemId = item.getId();
+            Object his = accessor_value.getValue(itemId);
+            if (!ObjectUtl.isEquals(itemState.getQuality(), QualityTypeValue.OPC_QUALITY_GOOD) && his != null) {
+              log.warn("opc 值不健康 item: {}, 状态: {}", itemId, itemState.getQuality());
+            }
+
+            if (!UnifiedDataAppService.isEquals(value, his)) {
+              OpcItemDto itemDto = this.getItem(itemId);
+              if (true) {
+                this.logItemChanged(itemId, accessor_value, value, itemDto);
+              }
+
+              accessor_value.setValue(itemId, value);
+            }
           }
+
+          end = System.currentTimeMillis();
+          if (log.isTraceEnabled()) {
+            log.trace("{}", itemsString);
+            log.trace("{} 计算完成耗时{}", tag, end - begin);
+          }
+
+          ThreadUtl.sleep((long) OpcConfig.synchronized_millisecond);
+          if (this.error_num != 0) {
+            this.error_num = 0;
+            this.message = null;
+          }
+
+          if (!valueAllNotNull) {
+            int random = (new Random()).nextInt(10) + 1;
+            random *= 1000;
+            if (this.all_null < 3) {
+              if (log.isWarnEnabled()) {
+                log.warn("{} 所有内容都为空, all_null:{} ,暂定{}s", tag, all_null,5000 + random);
+              }
+
+              ThreadUtl.sleep((long) (5000 + random));
+            } else if (this.all_null < 6) {
+              if (log.isWarnEnabled()) {
+                log.warn(tag + "重新创建server");
+                log.warn("{} 所有内容都为空, all_null:{} ,暂定{}s", tag, all_null,30000 + random);
+              }
+//                            ThreadUtl.sleep((long) (30000 + random));
+              ThreadUtl.sleep((long) ((new Random()).nextInt(3) +1) * 1000);
+              break start;
+            } else if (this.all_null < 12) {
+              if (log.isWarnEnabled()) {
+                log.warn("{} 所有内容都为空, all_null:{} ,暂定{}ms", tag, all_null, '\uea60' + random);
+              }
+
+              ThreadUtl.sleep((long) ('\uea60' + random));
+            } else {
+              if (log.isWarnEnabled()) {
+                log.warn("{} 所有内容都为空, all_null:{} ,暂定{}ms", tag, all_null, 120000 + random);
+              }
+
+              ThreadUtl.sleep((long) (120000 + random));
+            }
+
+            ++this.all_null;
+          } else {
+            this.all_null = 0;
+          }
+//                    break start;
+
         }
-      } catch (Exception var30) {
+
+        log.warn("opc线程停止。。。");
+        return;
+      } catch (Exception var27) {
         if (this.server != null) {
           try {
             this.server.disconnect();
           } catch (Exception var25) {
-            log.warn("{} : server  disconnect", var25);
           }
         }
 
         this.server = null;
+        if (!DeviceOpcSynchronizeAutoRun.isRun) {
+          log.warn("opc线程停止2。。。");
+          return;
+        }
 
-        String error_message = "设备信息同步异常,"+var30;
+        String error_message = "设备信息同步异常";
         if (!StrUtil.equals(this.message, error_message)) {
-          log.warn("", var30);
+          log.warn(error_message, var27);
         }
 
-        try {
+        ThreadUtl.sleep((long) (OpcConfig.synchronized_exception_wait_second * 1000));
-          Thread.sleep((long) (OpcConfig.synchronized_exception_wait_second * 1000));
-        } catch (InterruptedException e) {
-          e.printStackTrace();
-          //log.warn("OPC 读取线程读取异常{} :", e);
-//          Thread.currentThread().interrupt();不会真正停止线程
-        }
         ++this.error_num;
         if (this.error_num > 3 && !StrUtil.equals(this.message, error_message)) {
-          log.info("设备同步通信异常");
           this.message = error_message;
         }
       }
     }
   }
 
+
+  private void runNew() {
+    Async20Access accessor = null;
+
+    while (true) {
+      String opcGroupId = this.getOpcGroupID();
+
+      try {
+        if (this.server == null) {
+          this.server = OpcServerUtl.getServerWithOutException(this.OpcServer.getOpc_host(), this.OpcServer.getCls_id(), this.OpcServer.getUser(), this.OpcServer.getPassword(), this.OpcServer.getDomain());
+          this.server.addStateListener(this);
+          accessor = new Async20Access(this.server, OpcConfig.synchronized_millisecond, true);
+          Iterator var9 = this.protocols.iterator();
+
+          while (var9.hasNext()) {
+            OpcItemDto protocol = (OpcItemDto) var9.next();
+            String itemId = protocol.getItem_code();
+            accessor.addItem(itemId, this);
+          }
+
+          accessor.bind();
+          log.info("Async20Access bind {}", opcGroupId);
+        }
+
+        Thread.sleep((long) (OpcConfig.synchronized_exception_wait_second * 1000));
+      } catch (Exception var8) {
+        if (accessor != null) {
+          try {
+            log.warn("Async20Access unbind {}", opcGroupId);
+            accessor.unbind();
+          } catch (Exception var7) {
+            var7.printStackTrace();
+          }
+
+          accessor = null;
+        }
+
+        if (this.server != null) {
+          try {
+            this.server.disconnect();
+          } catch (Exception var6) {
+          }
+
+          this.server = null;
+        }
+
+        if (var8 instanceof InterruptedException) {
+          log.warn("OPC 同步线程(%s)被中断", opcGroupId);
+          return;
+        }
+
+        log.warn("设备信息同步异常", var8);
+        ThreadUtl.sleep((long) (OpcConfig.synchronized_exception_wait_second * 1000));
+        String error_message = var8.getMessage();
+        if (error_message == null) {
+          error_message = var8.toString();
+        }
+
+        ++this.error_num;
+        if (this.error_num > 3 && !StrUtil.equals(this.message, error_message)) {
+          this.message = error_message;
+        }
+      }
+    }
+  }
+
+
+  public void connectionStateChanged(boolean connected) {
+    if (!connected) {
+      this.server = null;
+    }
+
+    log.warn("opc server {} {}", this.getOpcGroupID(), connected ? "connected" : "disconnected");
+  }
+
+  private String getOpcGroupID() {
+    String var10000 = this.OpcServer.getOpc_code();
+    return var10000 + "(" + this.protocols.size() + " items)";
+  }
+
   public static String formatDuring(long mss) {
     long days = mss / 86400000L;
     long hours = mss % 86400000L / 3600000L;
@@ -239,4 +368,57 @@ public class DeviceOpcProtocolRunable implements Runnable {
     long seconds = mss % 60000L / 1000L;
     return days + " days " + hours + " hours " + minutes + " minutes " + seconds + " seconds ";
   }
+
+
+  public void changed(Item item, ItemState itemState) {
+    String itemId = item.getId();
+
+    try {
+      Object value = OpcUtl.getValue(item, itemState);
+      UnifiedDataAccessor accessor_value = UnifiedDataAccessorFactory.getAccessor(OpcConfig.udw_opc_value_key);
+      accessor_value.setValue(itemId, value);
+
+//            if (value != null) {
+//                if (log.isTraceEnabled()) {
+//                    log.trace("Item {} new value: {}, Timestamp: {}", new Object[]{itemId, itemState.getValue(), itemState.getTimestamp().getTime()});
+//                }
+//            } else if (log.isInfoEnabled()) {
+//                log.info("Item {} new value: {}, Timestamp: {}, Quality: {}", new Object[]{itemId, itemState.getValue(), itemState.getTimestamp().getTime(), itemState.getQuality()});
+//            }
+      log.trace("Item {} new value: {}, Timestamp: {}", new Object[]{itemId, itemState.getValue(), itemState.getTimestamp().getTime()});
+
+      OpcItemDto itemDto = this.getItem(itemId);
+//            if (Boolean.TRUE.equals(itemDto.getNeed_log())) {
+//                this.logItemChanged(itemId, accessor_value, value, itemDto);
+//            }
+      this.logItemChanged(itemId, accessor_value, value, itemDto);
+
+    } catch (Exception var7) {
+      log.error(itemId, var7);
+    }
+
+  }
+
+  private void logItemChanged(String itemId, UnifiedDataAccessor accessor_value, Object value, OpcItemDto itemDto) {
+    LuceneExecuteLogService lucene = SpringContextHolder.getBean(LuceneExecuteLogService.class);
+    Object his = accessor_value.getValue(itemId);
+    List<String> relate_items = itemDto.getRelate_items();
+    if (relate_items != null && !relate_items.isEmpty()) {
+      StringBuilder sb = new StringBuilder();
+      Iterator var8 = relate_items.iterator();
+
+      while (var8.hasNext()) {
+        String relate = (String) var8.next();
+        Object obj = accessor_value.getValue(relate);
+        sb.append("key:" + relate + "value:" + obj + ";");
+      }
+      lucene.deviceExecuteLog(new LuceneLogDto(itemDto.getOpc_server_code(), itemDto.getOpc_plc_code(), itemDto.getDevice_code(), itemDto.getItem_code().substring(itemDto.getItem_code().lastIndexOf(".")+1), String.valueOf(his), String.valueOf(value)));
+
+    } else {
+
+      lucene.deviceExecuteLog(new LuceneLogDto(itemDto.getOpc_server_code(), itemDto.getOpc_plc_code(), itemDto.getDevice_code(), itemDto.getItem_code().substring(itemDto.getItem_code().lastIndexOf(".")+1), String.valueOf(his), String.valueOf(value)));
+
+    }
+  }
+
 }