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 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 itemSearchCache; public DeviceOpcProtocolRunable() { this.error_num = 0; - this.all_null = 0; this.message = null; - this.itemSearchCache = new HashMap(); - this.server = null; } public List 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 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 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 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 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 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 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}); - } - } - }