From 289f959feec2ec6d70ad6e15bacaf1a0961b9d86 Mon Sep 17 00:00:00 2001
From: "USER-20220102CG\\noblelift" <546428999@qq.com>
Date: Mon, 18 Sep 2023 23:02:00 +0800
Subject: [PATCH] =?UTF-8?q?rev=20=E6=9B=B4=E6=96=B0=E8=80=81=E7=89=88?=
 =?UTF-8?q?=E7=BA=BF=E7=A8=8B=E8=AF=BB=E5=86=99=E9=97=AE=E9=A2=98?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 .../nl/acs/opc/DeviceOpcProtocolRunable.java  | 387 ++++--------------
 1 file changed, 90 insertions(+), 297 deletions(-)

diff --git a/acs/hd/nladmin-system/src/main/java/org/nl/acs/opc/DeviceOpcProtocolRunable.java b/acs/hd/nladmin-system/src/main/java/org/nl/acs/opc/DeviceOpcProtocolRunable.java
index 5817fb8..bda5cad 100644
--- a/acs/hd/nladmin-system/src/main/java/org/nl/acs/opc/DeviceOpcProtocolRunable.java
+++ b/acs/hd/nladmin-system/src/main/java/org/nl/acs/opc/DeviceOpcProtocolRunable.java
@@ -4,33 +4,27 @@ 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.nl.acs.udw.UnifiedDataAppService;
-import org.openscada.opc.lib.da.*;
+import org.nl.modules.udw.UnifiedDataAccessor;
+import org.nl.modules.udw.UnifiedDataAccessorFactory;
+import org.nl.utils.RedisUtils;
+import org.nl.utils.SpringContextHolder;
+import org.openscada.opc.lib.da.Group;
+import org.openscada.opc.lib.da.Item;
+import org.openscada.opc.lib.da.ItemState;
+import org.openscada.opc.lib.da.Server;
 
 import java.util.*;
 
-
 @Slf4j
-public class DeviceOpcProtocolRunable implements Runnable, DataCallback, ServerConnectionStateListener {
+public class DeviceOpcProtocolRunable implements Runnable {
     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;
     }
 
     public List<OpcItemDto> getProtocols() {
@@ -49,58 +43,32 @@ public class DeviceOpcProtocolRunable implements Runnable, DataCallback, ServerC
         this.OpcServer = opcServer;
     }
 
+    OpcItemDto getItem(String item) {
+        Iterator var2 = this.protocols.iterator();
 
-    private OpcItemDto getItem(String item) {
-        OpcItemDto x = (OpcItemDto) this.itemSearchCache.get(item);
-        if (x == 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;
-                }
+        OpcItemDto dto;
+        do {
+            if (!var2.hasNext()) {
+                return null;
             }
-        }
 
-        return x;
-    }
+            dto = (OpcItemDto) var2.next();
+        } while (!StrUtil.equals(item, dto.getItem_code()));
 
+        return dto;
+    }
 
     @Override
     public void run() {
-        if (OpcConfig.opc_item_read_using_callback) {
-            this.runNew();
-        } else {
-            this.runOld();
-        }
-    }
-
-
-    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();
+                Server server = OpcServerUtl.getServerWithOutException(this.OpcServer.getOpc_host(), this.OpcServer.getCls_id(), this.OpcServer.getUser(), this.OpcServer.getPassword(), this.OpcServer.getDomain());
+                Group group = server.addGroup();
                 List<String> itemsString = new ArrayList();
-                Iterator var3 = this.protocols.iterator();
+                Iterator it = this.protocols.iterator();
 
-                while (var3.hasNext()) {
-                    OpcItemDto protocol = (OpcItemDto) var3.next();
+                while (it.hasNext()) {
+                    OpcItemDto protocol = (OpcItemDto) it.next();
                     String item = protocol.getItem_code();
                     itemsString.add(item);
                 }
@@ -108,57 +76,47 @@ public class DeviceOpcProtocolRunable implements Runnable, DataCallback, ServerC
                 Map<String, Item> itemsMap = new LinkedHashMap();
                 boolean is_error = false;
                 StringBuilder err_message = new StringBuilder();
-                Iterator var6 = itemsString.iterator();
+                Iterator var7 = itemsString.iterator();
 
-                while (var6.hasNext()) {
-                    String string = (String) var6.next();
+                while (var7.hasNext()) {
+                    String string = (String) var7.next();
 
                     try {
-                        Item item = group.addItem(string);
-                        itemsMap.put(string, item);
+                        itemsMap.put(string, group.addItem(string));
                         log.trace("添加成功 {}", string);
-                    } catch (Exception var26) {
-                        err_message.append(string + ":" + var26.getMessage());
+                    } catch (Exception var29) {
+                        err_message.append(string + ":" + var29.getMessage());
                         if (!is_error) {
                             is_error = true;
                         }
                     }
                 }
 
-                String tag;
                 if (is_error) {
-                    tag = err_message.toString();
-                    log.warn("{}:{}", OpcConfig.resource_code, tag);
+                    log.info("设备OPC数据同步配置异常");
                 }
 
                 if (!OpcStartTag.is_run) {
                     OpcStartTag.is_run = true;
                 }
 
-                tag = "";
-                if (log.isWarnEnabled()) {
-                    tag = Thread.currentThread().getName();
-                    if (this.OpcServer != null) {
-                        tag = tag + this.getOpcGroupID();
-                    }
+                //线程名
+                String tag = Thread.currentThread().getName();
+                if (this.OpcServer != null) {
+                    tag = tag + this.OpcServer.getOpc_code();
                 }
-
                 UnifiedDataAccessor accessor_value = UnifiedDataAccessorFactory.getAccessor(OpcConfig.udw_opc_value_key);
+
                 boolean time_out = false;
 
-                while (DeviceOpcSynchronizeAutoRun.isRun) {
+                label97:
+                while (true) {
                     long begin = System.currentTimeMillis();
-                    if (log.isTraceEnabled()) {
-                        log.trace("{} 开始记时{}", tag, DateUtil.now());
-                    }
-
                     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;
-                    if (log.isTraceEnabled()) {
-                        log.trace("{} 读取耗时:{}", tag, duration);
-                    }
-
+                    log.trace("{} 读取耗时:{}", tag, duration);
                     if (duration > 1000L) {
                         if (!time_out) {
                             log.warn("{} 读取超时 : {}", tag, duration);
@@ -169,198 +127,84 @@ public class DeviceOpcProtocolRunable implements Runnable, DataCallback, ServerC
                         time_out = false;
                     }
 
-                    boolean valueAllNotNull = false;
                     Set<Item> items = itemStatus.keySet();
                     Iterator var18 = items.iterator();
 
-                    while (var18.hasNext()) {
-                        Item item = (Item) var18.next();
-                        ItemState itemState = (ItemState) itemStatus.get(item);
-                        Object value = OpcUtl.getValue(item, itemState);
-                        if (value != null) {
-                            valueAllNotNull = true;
-                        }
-
-                        String itemId = item.getId();
-                        Object his = accessor_value.getValue(itemId);
-                        if (!ObjectUtl.isEquals(itemState.getQuality(), QualityTypeValue.OPC_QUALITY_GOOD) && his != null) {
-//                            if (itemState.getQuality()==QualityTypeValue.OPC_QUALITY_GOOD && his != null) {
-                            log.warn("opc 值不健康 item: {}, 状态: {}", itemId, itemState.getQuality());
-                            valueAllNotNull = false;
-                        }
-
-                        if (!UnifiedDataAppService.isEquals(value, his)) {
-                            OpcItemDto itemDto = this.getItem(itemId);
-                            if (true) {
-                                this.logItemChanged(itemId, accessor_value, value, itemDto);
-                            }
-                            if(!ObjectUtil.isEmpty(value)){
-                                accessor_value.setValue(itemId, value);
+                    while (true) {
+                        Item item;
+                        //当前值
+                        Object value;
+                        //旧的值
+                        Object his;
+                        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;
                             }
-                        }
-                    }
-
-                    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;
-                    }
+                            item = (Item) var18.next();
+                            ItemState itemState = (ItemState) itemStatus.get(item);
+                            value = OpcUtl.getValue(item, itemState);
+                            his = accessor_value.getValue(item.getId());
+//                            RedisUtils redisUtils = SpringContextHolder.getBean(RedisUtils.class);
+//                            his=redisUtils.get(item.getId());
 
-                    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) ((new Random()).nextInt(3) +1) * 1000);
-                            break start;
-                        } else if (this.all_null < 6) {
-                            if (log.isWarnEnabled()) {
-                                log.warn(tag + "重新创建server");
-                                log.warn("{} 所有内容都为空, all_null:{} ,暂定{}s", tag, all_null,30000 + random);
+                            if (!ObjectUtil.equal(itemState.getQuality(), QualityTypeValue.OPC_QUALITY_GOOD) && his != null) {
+                                log.warn("opc 值不健康 item: {}, 状态: {}", item.getId(), itemState.getQuality());
                             }
-//                            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);
+                        } while (ObjectUtil.equal(value, his));//如果两次的值相等,不走下面的代码
+
+                        OpcItemDto itemDto = this.getItem(item.getId());
+                        if (itemDto.getNeed_log() != null && itemDto.getNeed_log()) {
+                            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 + ";");
                             }
 
-                            ThreadUtl.sleep((long) ('\uea60' + random));
-                        } else {
-                            if (log.isWarnEnabled()) {
-                                log.warn("{} 所有内容都为空, all_null:{} ,暂定{}ms", tag, all_null, 120000 + random);
-                            }
+                            log.info("信号{}变更从{}->{};信号快照:{}", new Object[]{item.getId(), his, value, sb});
 
-                            ThreadUtl.sleep((long) (120000 + random));
                         }
+                        //设置值
+                        accessor_value.setValue(item.getId(), value);
+                        RedisUtils redisUtils = SpringContextHolder.getBean(RedisUtils.class);
+                        redisUtils.set(item.getId(), value);
 
-                        ++this.all_null;
-                    } else {
-                        this.all_null = 0;
-                    }
-//                    break start;
 
-                }
-
-                log.warn("opc线程停止。。。");
-                return;
-            } catch (Exception var27) {
-                if (this.server != null) {
-                    try {
-                        this.server.disconnect();
-                    } catch (Exception var25) {
                     }
                 }
-
-                this.server = null;
-                if (!DeviceOpcSynchronizeAutoRun.isRun) {
-                    log.warn("opc线程停止2。。。");
-                    return;
-                }
-
+            } catch (Exception var30) {
                 String error_message = "设备信息同步异常";
                 if (!StrUtil.equals(this.message, error_message)) {
-                    log.warn(error_message, var27);
+                    log.warn("", var30);
                 }
 
-                ThreadUtl.sleep((long) (OpcConfig.synchronized_exception_wait_second * 1000));
-                ++this.error_num;
-                if (this.error_num > 3 && !StrUtil.equals(this.message, error_message)) {
-                    this.message = error_message;
+                try {
+                    Thread.sleep((long) (OpcConfig.synchronized_exception_wait_second * 1000));
+                } catch (InterruptedException e) {
+                    e.printStackTrace();
                 }
-            }
-        }
-    }
-
-
-    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)) {
+                    log.info("设备同步通信异常");
                     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;
@@ -368,55 +212,4 @@ public class DeviceOpcProtocolRunable implements Runnable, DataCallback, ServerC
         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) {
-        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 + ";");
-            }
-            log.warn("设备:{}信号{}变更从{}->{};信号快照:{}", new Object[]{itemDto.getDevice_code(), itemId, his, value, sb});
-//            this.businessLogger.setResource(itemDto.getDevice_code(), itemDto.getDevice_name()).log("信号{}变更从{}->{};信号快照:{}", new Object[]{itemId, his, value, sb});
-        } else {
-            log.warn("设备:{}信号{}变更从{}->{};信号快照:{}", new Object[]{itemDto.getDevice_code(), itemId, his, value});
-//            this.businessLogger.setResource(itemDto.getDevice_code(), itemDto.getDevice_name()).log("信号{}变更从{}->{}", new Object[]{itemId, his, value});
-        }
-    }
-
 }