wangs
/
maGang
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity
Browse Source

更新

master
gengby 2 years ago
parent
b428f0686e
commit
8373cb9b5b
2 changed files with 286 additions and 160 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Unified View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 390
      acs/nladmin-system/src/main/java/org/nl/acs/opc/DeviceOpcProtocolRunable.java
  2. 2
      acs/nladmin-system/src/main/resources/config/application.yml

390
acs/nladmin-system/src/main/java/org/nl/acs/opc/DeviceOpcProtocolRunable.java
View File

@ -1,30 +1,39 @@
package org.nl.acs.opc; package org.nl.acs.opc;
import cn.hutool.core.date.DateUtil;
import cn.hutool.core.util.ObjectUtil; import cn.hutool.core.util.ObjectUtil;
import cn.hutool.core.util.StrUtil; import com.alibaba.fastjson.JSON;
import lombok.extern.slf4j.Slf4j; 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.UnifiedDataAccessor;
import org.nl.acs.udw.UnifiedDataAccessorFactory; 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.common.exception.BadRequestException;
import org.openscada.opc.lib.da.ItemState; import org.nl.modules.wql.util.SpringContextHolder;
import org.openscada.opc.lib.da.Server; import org.openscada.opc.lib.da.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.*; import java.util.*;
@Slf4j public class DeviceOpcProtocolRunable implements Runnable, DataCallback, ServerConnectionStateListener {
public class DeviceOpcProtocolRunable implements Runnable { private static final Logger log = LoggerFactory.getLogger(DeviceOpcProtocolRunable.class);
List<OpcItemDto> protocols; private List<OpcItemDto> protocols;
OpcServerManageDto OpcServer; private OpcServerManageDto opcServer;
int error_num; private int error_num;
int all_null; private int all_null;
String message; private String message;
private Map<String, OpcItemDto> itemSearchCache;
private Server server;
public DeviceOpcProtocolRunable() { public DeviceOpcProtocolRunable() {
this.error_num = 0; this.error_num = 0;
this.all_null = 0; this.all_null = 0;
this.message = null; this.message = null;
this.itemSearchCache = new HashMap();
this.server = null;
} }
public List<OpcItemDto> getProtocols() { public List<OpcItemDto> getProtocols() {
@ -36,40 +45,117 @@ public class DeviceOpcProtocolRunable implements Runnable {
} }
public OpcServerManageDto getOpcServer() { public OpcServerManageDto getOpcServer() {
return this.OpcServer; return this.opcServer;
} }
public void setOpcServer(OpcServerManageDto opcServer) { public void setOpcServer(OpcServerManageDto opcServer) {
this.OpcServer = opcServer; this.opcServer = opcServer;
} }
OpcItemDto getItem(String item) { private OpcItemDto getItem(String item) {
Iterator var2 = this.protocols.iterator(); OpcItemDto x = (OpcItemDto) this.itemSearchCache.get(item);
if (x == null) {
Iterator var3 = this.protocols.iterator();
OpcItemDto dto; while (var3.hasNext()) {
do { OpcItemDto dto = (OpcItemDto) var3.next();
if (!var2.hasNext()) { if (ObjectUtil.equals(item, dto.getItem_code())) {
return null; 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() { 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) {
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, 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) { while (true) {
Server server = null;
try { try {
server = OpcServerUtl.getServerWithOutException(this.OpcServer.getOpc_host(), this.OpcServer.getCls_id(), this.OpcServer.getUser(), this.OpcServer.getPassword(), this.OpcServer.getDomain()); this.server = OpcServerUtl.getServerWithOutException(this.opcServer.getOpc_host(), this.opcServer.getCls_id(), this.opcServer.getUser(), this.opcServer.getPassword(), this.opcServer.getDomain());
Group group = server.addGroup(); this.server.addStateListener(this);
Group group = this.server.addGroup();
List<String> itemsString = new ArrayList(); List<String> itemsString = new ArrayList();
Iterator it = this.protocols.iterator(); Iterator var3 = this.protocols.iterator();
while (it.hasNext()) { while (var3.hasNext()) {
OpcItemDto protocol = (OpcItemDto) it.next(); OpcItemDto protocol = (OpcItemDto) var3.next();
String item = protocol.getItem_code(); String item = protocol.getItem_code();
itemsString.add(item); itemsString.add(item);
} }
@ -77,49 +163,57 @@ public class DeviceOpcProtocolRunable implements Runnable {
Map<String, Item> itemsMap = new LinkedHashMap(); Map<String, Item> itemsMap = new LinkedHashMap();
boolean is_error = false; boolean is_error = false;
StringBuilder err_message = new StringBuilder(); 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 { try {
itemsMap.put(string, group.addItem(string)); Item item = group.addItem(string);
log.info("添加成功 {}", string); itemsMap.put(string, item);
} catch (Exception var29) { log.trace("添加成功 {}", string);
err_message.append(string + ":" + var29.getMessage()); } catch (Exception var26) {
err_message.append(string + ":" + var26.getMessage());
if (!is_error) { if (!is_error) {
is_error = true; is_error = true;
} }
} }
} }
String tag;
if (is_error) { if (is_error) {
log.info("设备OPC数据同步配置异常"); tag = err_message.toString();
log.warn("{}:{}", OpcConfig.resource_code, tag);
} }
if (!OpcStartTag.is_run) { if (!OpcStartTag.is_run) {
OpcStartTag.is_run = true; 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; boolean time_out = false;
label97:
while (DeviceOpcSynchronizeAutoRun.isRun) { while (DeviceOpcSynchronizeAutoRun.isRun) {
log.info("label97_123");
long begin = System.currentTimeMillis(); long begin = System.currentTimeMillis();
if (log.isTraceEnabled()) {
// log.trace("{} 开始计时{}", tag, DateUtl.formatDateTime(new Date()));
}
Map<Item, ItemState> itemStatus = group.read(true, (Item[]) itemsMap.values().toArray(new Item[0])); Map<Item, ItemState> itemStatus = group.read(true, (Item[]) itemsMap.values().toArray(new Item[0]));
log.info("{}读取到的itemStatus结果为:", itemStatus);
long end = System.currentTimeMillis(); long end = System.currentTimeMillis();
log.info("{} 开始记时{}", tag, DateUtil.now());
long duration = end - begin; long duration = end - begin;
log.info("{} 读取耗时：{}", tag, duration); if (log.isTraceEnabled()) {
// log.trace("{} 读取耗时：{}", tag, DateUtl.formatDuring(duration));
}
if (duration > 1000L) { if (duration > 1000L) {
if (!time_out) { if (!time_out) {
log.warn("{} 读取超时 : {}", tag, duration); log.warn("{} 读取超时 : {}", tag, duration);
@ -129,41 +223,79 @@ public class DeviceOpcProtocolRunable implements Runnable {
} else { } else {
time_out = false; time_out = false;
} }
boolean valueAllNotNull = false; boolean valueAllNotNull = false;
Set<Item> items = itemStatus.keySet(); Set<Item> items = itemStatus.keySet();
log.info("items值为:", items);
Iterator var18 = items.iterator();
while (true) { Iterator var18 = items.iterator();
Item item;
//当前值
Object value;
//旧的值
Object his;
do {
if (!var18.hasNext()){ 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("查询到的值为空");
}
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);
}
accessor_value.setValue(itemId, value);
}
}
end = System.currentTimeMillis(); end = System.currentTimeMillis();
log.info("{}", itemsString); if (log.isTraceEnabled()) {
log.info("{} 计算完成耗时{}", tag, end - begin); log.trace("{}", itemsString);
Thread.sleep((long) OpcConfig.synchronized_millisecond); // log.trace("{} 计算完成耗时{}", tag, DateUtl.formatDuring(end - begin));
}
ThreadUtl.sleep((long) OpcConfig.synchronized_millisecond);
if (this.error_num != 0) { if (this.error_num != 0) {
this.error_num = 0; this.error_num = 0;
this.message = null; this.message = null;
} }
if (!valueAllNotNull) { if (!valueAllNotNull) {
int random = (new Random()).nextInt(10) + 1; int random = (new Random()).nextInt(10) + 1;
random *= 1000; random *= 1000;
if (this.all_null < 3) { if (this.all_null < 3) {
if (log.isWarnEnabled()) {
log.warn("{} 所有内容都为空,暂定{}s", tag, 5000 + random); log.warn("{} 所有内容都为空,暂定{}s", tag, 5000 + random);
}
ThreadUtl.sleep((long) (5000 + random)); ThreadUtl.sleep((long) (5000 + random));
} else if (this.all_null < 6) { } else if (this.all_null < 6) {
if (log.isWarnEnabled()) {
log.warn("{} 所有内容都为空,暂定{}s", tag, 30000 + random); log.warn("{} 所有内容都为空,暂定{}s", tag, 30000 + random);
}
ThreadUtl.sleep((long) (30000 + random)); ThreadUtl.sleep((long) (30000 + random));
} else if (this.all_null < 12) { } else if (this.all_null < 12) {
if (log.isWarnEnabled()) {
log.warn("{} 所有内容都为空,暂定{}ms", tag, '\uea60' + random); log.warn("{} 所有内容都为空,暂定{}ms", tag, '\uea60' + random);
}
ThreadUtl.sleep((long) ('\uea60' + random)); ThreadUtl.sleep((long) ('\uea60' + random));
} else { } else {
if (log.isWarnEnabled()) {
log.warn("{} 所有内容都为空,暂定{}ms", tag, 120000 + random); log.warn("{} 所有内容都为空,暂定{}ms", tag, 120000 + random);
}
ThreadUtl.sleep((long) (120000 + random)); ThreadUtl.sleep((long) (120000 + random));
} }
@ -171,98 +303,92 @@ public class DeviceOpcProtocolRunable implements Runnable {
} else { } else {
this.all_null = 0; this.all_null = 0;
} }
log.info("label97");
continue label97;
} }
log.info("label97A----");
item = (Item) var18.next(); log.warn("opc线程停止。。。");
log.info("读取到item:{}", item); return;
ItemState itemState = (ItemState) itemStatus.get(item); } catch (Exception var27) {
log.info("读取到itemState:{}", itemState); if (this.server != null) {
log.info("开始根据item,itemState读取信号值:{},{}", item, itemState); try {
value = OpcUtl.getValue(item, itemState); this.server.disconnect();
log.info("根据item，itemState读取到信号值为:{}", value); } catch (Exception var25) {
if (value != null) {
valueAllNotNull = true;
} }
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());
} }
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()); this.server = null;
// if (itemDto.getNeed_log() != null && itemDto.getNeed_log()) { if (!DeviceOpcSynchronizeAutoRun.isRun) {
if (true) { log.warn("opc线程停止2。。。");
StringBuilder sb = new StringBuilder(); return;
//设备的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.info("信号{}变更从{}->{};信号快照：{}", new Object[]{item.getId(), his, value, sb}); // String error_message = MessageFormatUtl.format("设备信息同步异常：{}", new Object[]{var27.getMessage()});
String error_message = "";
if (!ObjectUtil.equals(this.message, error_message)) {
log.warn(error_message, var27);
}
ThreadUtl.sleep((long) (OpcConfig.synchronized_exception_wait_second * 1000));
++this.error_num;
if (this.error_num > 3 && !ObjectUtil.equals(this.message, error_message)) {
this.message = error_message;
} }
log.info("设置值");
//设置值
accessor_value.setValue(item.getId(), value);
} }
} }
log.info("opc线程停止。。。");
return;
} catch (Exception var30) {
if (!DeviceOpcSynchronizeAutoRun.isRun) {
log.info("opc线程停止2。。。");
if (server != null) {
try {
server.disconnect();
} catch (Exception var31) {
} }
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()});
} }
return; } else if (log.isInfoEnabled()) {
log.info("Item {} new value: {}, Timestamp: {}, Quality: {}", new Object[]{itemId, itemState.getValue(), itemState.getTimestamp().getTime(), itemState.getQuality()});
} }
String error_message = "设备信息同步异常"; OpcItemDto itemDto = this.getItem(itemId);
if (!StrUtil.equals(this.message, error_message)) { if (Boolean.TRUE.equals(itemDto.getNeed_log())) {
log.warn("", var30); this.logItemChanged(itemId, accessor_value, value, itemDto);
} }
} catch (BadRequestException var7) {
if (server != null) { log.error(itemId, var7);
try {
server.disconnect();
} catch (Exception e) {
} }
} }
try { private void logItemChanged(String itemId, UnifiedDataAccessor accessor_value, Object value, OpcItemDto itemDto) {
Thread.sleep((long) (OpcConfig.synchronized_exception_wait_second * 1000)); Object his = accessor_value.getValue(itemId);
} catch (InterruptedException e) { List<String> relate_items = itemDto.getRelate_items();
e.printStackTrace(); if (relate_items != null && !relate_items.isEmpty()) {
Thread.currentThread().interrupt(); 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 + ";");
} }
++this.error_num;
if (this.error_num > 3 && !StrUtil.equals(this.message, error_message)) { } else {
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");
} }
public static String formatDuring(long mss) { private String getOpcGroupID() {
long days = mss / 86400000L; return this.opcServer.getOpc_code() + "(" + this.protocols.size() + " items)";
long hours = mss % 86400000L / 3600000L;
long minutes = mss % 3600000L / 60000L;
long seconds = mss % 60000L / 1000L;
return days + " days " + hours + " hours " + minutes + " minutes " + seconds + " seconds ";
} }
} }

2
acs/nladmin-system/src/main/resources/config/application.yml
View File

@ -2,7 +2,7 @@ spring:
freemarker: freemarker:
check-template-location: false check-template-location: false
profiles: profiles:
active: prod active: dev
jackson: jackson:
time-zone: GMT+8 time-zone: GMT+8
data: data:

Write Preview
Loading…
Cancel
Save