From 8373cb9b5bc2e66bda13b49a3bb587e8ec9a185b Mon Sep 17 00:00:00 2001 From: gengby <858962040@qq.com> Date: Thu, 12 Jan 2023 13:17:21 +0800 Subject: [PATCH] =?UTF-8?q?=E6=9B=B4=E6=96=B0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../nl/acs/opc/DeviceOpcProtocolRunable.java | 444 +++++++++++------- .../src/main/resources/config/application.yml | 2 +- 2 files changed, 286 insertions(+), 160 deletions(-) diff --git a/acs/nladmin-system/src/main/java/org/nl/acs/opc/DeviceOpcProtocolRunable.java b/acs/nladmin-system/src/main/java/org/nl/acs/opc/DeviceOpcProtocolRunable.java index 0b9190f..d3b12ec 100644 --- a/acs/nladmin-system/src/main/java/org/nl/acs/opc/DeviceOpcProtocolRunable.java +++ b/acs/nladmin-system/src/main/java/org/nl/acs/opc/DeviceOpcProtocolRunable.java @@ -1,30 +1,39 @@ + 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 com.alibaba.fastjson.JSON; +import org.nl.acs.auto.run.AutoRunService; +import org.nl.acs.auto.run.AutoRunServiceImpl; import org.nl.acs.udw.UnifiedDataAccessor; import org.nl.acs.udw.UnifiedDataAccessorFactory; -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 org.nl.acs.udw.UnifiedDataAppService; +import org.nl.modules.common.exception.BadRequestException; +import org.nl.modules.wql.util.SpringContextHolder; +import org.openscada.opc.lib.da.*; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; import java.util.*; -@Slf4j -public class DeviceOpcProtocolRunable implements Runnable { - List protocols; - OpcServerManageDto OpcServer; - int error_num; - int all_null; - String message; +public class DeviceOpcProtocolRunable implements Runnable, DataCallback, ServerConnectionStateListener { + private static final Logger log = LoggerFactory.getLogger(DeviceOpcProtocolRunable.class); + private List protocols; + private OpcServerManageDto opcServer; + private int error_num; + private int all_null; + private String message; + private Map itemSearchCache; + private Server server; + public DeviceOpcProtocolRunable() { this.error_num = 0; this.all_null = 0; this.message = null; + this.itemSearchCache = new HashMap(); + this.server = null; } public List getProtocols() { @@ -36,40 +45,117 @@ public class DeviceOpcProtocolRunable implements Runnable { } public OpcServerManageDto getOpcServer() { - return this.OpcServer; + return this.opcServer; } public void setOpcServer(OpcServerManageDto opcServer) { - this.OpcServer = opcServer; + this.opcServer = opcServer; } - OpcItemDto getItem(String item) { - Iterator var2 = this.protocols.iterator(); - - OpcItemDto dto; - do { - if (!var2.hasNext()) { - return null; + 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 (ObjectUtil.equals(item, dto.getItem_code())) { + x = dto; + this.itemSearchCache.put(item, dto); + break; + } } + } - dto = (OpcItemDto) var2.next(); - } while (!StrUtil.equals(item, dto.getItem_code())); - - return dto; + return x; } - @Override public void run() { + /* if (OpcConfig.opc_item_read_using_callback) { + this.runNew(); + } else { + this.runOld(); + } +*/ + this.runOld(); + } + + private void runNew() { + Async20Access accessor = null; + while (true) { - Server server = null; + String opcGroupId = this.getOpcGroupID(); + try { - server = OpcServerUtl.getServerWithOutException(this.OpcServer.getOpc_host(), this.OpcServer.getCls_id(), this.OpcServer.getUser(), this.OpcServer.getPassword(), this.OpcServer.getDomain()); - Group group = server.addGroup(); + 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, false); + 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 && !ObjectUtil.equal(this.message, error_message)) { + this.message = error_message; + } + } + } + } + + private void runOld() { + while (true) { + try { + 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 group = this.server.addGroup(); List itemsString = new ArrayList(); - Iterator it = this.protocols.iterator(); + Iterator var3 = this.protocols.iterator(); - while (it.hasNext()) { - OpcItemDto protocol = (OpcItemDto) it.next(); + while (var3.hasNext()) { + OpcItemDto protocol = (OpcItemDto) var3.next(); String item = protocol.getItem_code(); itemsString.add(item); } @@ -77,49 +163,57 @@ public class DeviceOpcProtocolRunable implements Runnable { Map itemsMap = new LinkedHashMap(); boolean is_error = false; StringBuilder err_message = new StringBuilder(); - Iterator var7 = itemsString.iterator(); + Iterator var6 = itemsString.iterator(); - while (var7.hasNext()) { - String string = (String) var7.next(); + while (var6.hasNext()) { + String string = (String) var6.next(); try { - itemsMap.put(string, group.addItem(string)); - log.info("添加成功 {}", string); - } catch (Exception var29) { - err_message.append(string + ":" + var29.getMessage()); + Item item = group.addItem(string); + itemsMap.put(string, item); + log.trace("添加成功 {}", string); + } catch (Exception var26) { + 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; } - //线程名 - String tag = Thread.currentThread().getName(); - if (this.OpcServer != null) { - tag = tag + this.OpcServer.getOpc_code(); + tag = ""; + if (log.isWarnEnabled()) { + tag = Thread.currentThread().getName(); + 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) { - log.info("label97_123"); long begin = System.currentTimeMillis(); + if (log.isTraceEnabled()) { +// log.trace("{} 开始计时{}", tag, DateUtl.formatDateTime(new Date())); + } + Map itemStatus = group.read(true, (Item[]) itemsMap.values().toArray(new Item[0])); - log.info("{}读取到的itemStatus结果为:", itemStatus); long end = System.currentTimeMillis(); - log.info("{} 开始记时{}", tag, DateUtil.now()); long duration = end - begin; - log.info("{} 读取耗时：{}", tag, duration); + if (log.isTraceEnabled()) { +// log.trace("{} 读取耗时：{}", tag, DateUtl.formatDuring(duration)); + } + if (duration > 1000L) { if (!time_out) { log.warn("{} 读取超时 : {}", tag, duration); @@ -129,140 +223,172 @@ public class DeviceOpcProtocolRunable implements Runnable { } else { time_out = false; } + boolean valueAllNotNull = false; Set items = itemStatus.keySet(); - log.info("items值为:", items); + Iterator var18 = items.iterator(); + if (!var18.hasNext()){ + log.info("查询到的为空！"); + } + + 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; + }else { + log.info("查询到的值为空"); + } - while (true) { - Item item; - //当前值 - Object value; - //旧的值 - Object his; - do { - if (!var18.hasNext()) { - end = System.currentTimeMillis(); - log.info("{}", itemsString); - log.info("{} 计算完成耗时{}", tag, end - begin); - Thread.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) { - log.warn("{} 所有内容都为空,暂定{}s", tag, 5000 + random); - ThreadUtl.sleep((long) (5000 + random)); - } else if (this.all_null < 6) { - log.warn("{} 所有内容都为空,暂定{}s", tag, 30000 + random); - ThreadUtl.sleep((long) (30000 + random)); - } else if (this.all_null < 12) { - log.warn("{} 所有内容都为空,暂定{}ms", tag, '\uea60' + random); - ThreadUtl.sleep((long) ('\uea60' + random)); - } else { - log.warn("{} 所有内容都为空,暂定{}ms", tag, 120000 + random); - ThreadUtl.sleep((long) (120000 + random)); - } - - ++this.all_null; - } else { - this.all_null = 0; - } - log.info("label97"); - continue label97; + String itemId = item.getId(); + Object his = accessor_value.getValue(itemId); + if (!ObjectUtil.equals(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 (Boolean.TRUE.equals(itemDto.getNeed_log())) { + this.logItemChanged(itemId, accessor_value, value, itemDto); } - log.info("label97A----"); - item = (Item) var18.next(); - log.info("读取到item:{}", item); - ItemState itemState = (ItemState) itemStatus.get(item); - log.info("读取到itemState:{}", itemState); - log.info("开始根据item,itemState读取信号值:{},{}", item, itemState); - value = OpcUtl.getValue(item, itemState); - log.info("根据item，itemState读取到信号值为:{}", value); - if (value != null) { - valueAllNotNull = true; + + accessor_value.setValue(itemId, value); + } + } + + end = System.currentTimeMillis(); + if (log.isTraceEnabled()) { + log.trace("{}", itemsString); +// log.trace("{} 计算完成耗时{}", tag, DateUtl.formatDuring(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("{} 所有内容都为空,暂定{}s", tag, 5000 + random); } - his = accessor_value.getValue(item.getId()); - log.info("读取到旧的信号值:{}", his); - if (!ObjectUtil.equal(itemState.getQuality(), QualityTypeValue.OPC_QUALITY_GOOD) && his != null) { - log.warn("opc 值不健康 item: {}, 状态: {}", item.getId(), itemState.getQuality()); + + ThreadUtl.sleep((long) (5000 + random)); + } else if (this.all_null < 6) { + if (log.isWarnEnabled()) { + log.warn("{} 所有内容都为空,暂定{}s", tag, 30000 + random); } - log.info("label97bbb----value{},his{},equals{}",value,his,ObjectUtil.equal(value, his)); - } while (ObjectUtil.equal(value, his));//如果两次的值相等，不走下面的代码 - - log.info("如果两次的值相等，不走下面的代码"); - - OpcItemDto itemDto = this.getItem(item.getId()); -// if (itemDto.getNeed_log() != null && itemDto.getNeed_log()) { - if (true) { - 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) (30000 + random)); + } else if (this.all_null < 12) { + if (log.isWarnEnabled()) { + log.warn("{} 所有内容都为空,暂定{}ms", tag, '\uea60' + random); } - log.info("信号{}变更从{}->{};信号快照：{}", new Object[]{item.getId(), his, value, sb}); + ThreadUtl.sleep((long) ('\uea60' + random)); + } else { + if (log.isWarnEnabled()) { + log.warn("{} 所有内容都为空,暂定{}ms", tag, 120000 + random); + } + ThreadUtl.sleep((long) (120000 + random)); } - log.info("设置值"); - //设置值 - accessor_value.setValue(item.getId(), value); + + ++this.all_null; + } else { + this.all_null = 0; } } - log.info("opc线程停止。。。"); + + log.warn("opc线程停止。。。"); return; - } catch (Exception var30) { - if (!DeviceOpcSynchronizeAutoRun.isRun) { - log.info("opc线程停止2。。。"); - if (server != null) { - try { - server.disconnect(); - } catch (Exception var31) { - } + } catch (Exception var27) { + if (this.server != null) { + try { + this.server.disconnect(); + } catch (Exception var25) { } - return; } - String error_message = "设备信息同步异常"; - if (!StrUtil.equals(this.message, error_message)) { - log.warn("", var30); + this.server = null; + if (!DeviceOpcSynchronizeAutoRun.isRun) { + log.warn("opc线程停止2。。。"); + return; } - if (server != null) { - try { - server.disconnect(); - } catch (Exception e) { - } +// String error_message = MessageFormatUtl.format("设备信息同步异常：{}", new Object[]{var27.getMessage()}); + String error_message = ""; + if (!ObjectUtil.equals(this.message, error_message)) { + log.warn(error_message, var27); } - try { - Thread.sleep((long) (OpcConfig.synchronized_exception_wait_second * 1000)); - } catch (InterruptedException e) { - e.printStackTrace(); - Thread.currentThread().interrupt(); - } + ThreadUtl.sleep((long) (OpcConfig.synchronized_exception_wait_second * 1000)); ++this.error_num; - if (this.error_num > 3 && !StrUtil.equals(this.message, error_message)) { - log.info("设备同步通信异常"); + if (this.error_num > 3 && !ObjectUtil.equals(this.message, error_message)) { this.message = error_message; } } } } - public static String formatDuring(long mss) { - long days = mss / 86400000L; - long hours = mss % 86400000L / 3600000L; - long minutes = mss % 3600000L / 60000L; - 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()}); + } + + OpcItemDto itemDto = this.getItem(itemId); + if (Boolean.TRUE.equals(itemDto.getNeed_log())) { + this.logItemChanged(itemId, accessor_value, value, itemDto); + } + } catch (BadRequestException 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 + ";"); + } + + } else { + + } + + } + + public void connectionStateChanged(boolean connected) { + if (!connected) { + this.server = null; + } + + log.warn("opc server {} {}", this.getOpcGroupID(), connected ? "connected" : "disconnected"); + } + + private String getOpcGroupID() { + return this.opcServer.getOpc_code() + "(" + this.protocols.size() + " items)"; } } diff --git a/acs/nladmin-system/src/main/resources/config/application.yml b/acs/nladmin-system/src/main/resources/config/application.yml index a4d00ab..7e3fbea 100644 --- a/acs/nladmin-system/src/main/resources/config/application.yml +++ b/acs/nladmin-system/src/main/resources/config/application.yml @@ -2,7 +2,7 @@ spring: freemarker: check-template-location: false profiles: - active: prod + active: dev jackson: time-zone: GMT+8 data: