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

fix：opc线程

master
gengby 1 year ago
parent
5d7df2bf7d
commit
cbc53fbda1
5 changed files with 395 additions and 86 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 378
      hd/nladmin-system/src/main/java/org/nl/acs/opc/DeviceOpcProtocolRunable.java
  2. 7
      hd/nladmin-system/src/main/java/org/nl/acs/opc/DeviceOpcSynchronizeAutoRun.java
  3. 73
      hd/nladmin-system/src/main/java/org/nl/acs/opc/ObjectUtl.java
  4. 2
      hd/nladmin-system/src/main/java/org/nl/acs/opc/OpcConfig.java
  5. 21
      hd/nladmin-system/src/main/java/org/nl/acs/opc/ThreadUtl.java

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

@ -1,28 +1,35 @@
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.modules.udw.UnifiedDataAccessor;
import org.nl.modules.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.modules.udw.UnifiedDataAppService;
import org.openscada.opc.lib.da.*;
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;
}
public List<OpcItemDto> getProtocols() {
@ -41,32 +48,58 @@ public class DeviceOpcProtocolRunable implements Runnable {
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();
dto = (OpcItemDto) var2.next();
} while (!StrUtil.equals(item, dto.getItem_code()));
while (var3.hasNext()) {
OpcItemDto dto = (OpcItemDto) var3.next();
if (StrUtil.equals(item, dto.getItem_code())) {
x = dto;
this.itemSearchCache.put(item, dto);
break;
}
}
}
return dto;
return x;
}
@Override
public void run() {
if (OpcConfig.opc_item_read_using_callback) {
this.runNew();
} else {
this.runOld();
}
}
private void runOld() {
while (true) {
start:
try {
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();
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 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);
}
@ -74,47 +107,57 @@ 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()) {
String string = (String) var7.next();
while (var6.hasNext()) {
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) {
err_message.append(string + ":" + var29.getMessage());
} 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 (true) {
while (DeviceOpcSynchronizeAutoRun.isRun) {
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;
log.trace("{} 读取耗时：{}", tag, duration);
if (log.isTraceEnabled()) {
log.trace("{} 读取耗时：{}", tag, duration);
}
if (duration > 1000L) {
if (!time_out) {
log.warn("{} 读取超时 : {}", tag, duration);
@ -125,78 +168,196 @@ public class DeviceOpcProtocolRunable implements Runnable {
time_out = false;
}
boolean valueAllNotNull = false;
Set<Item> items = itemStatus.keySet();
Iterator var18 = items.iterator();
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;
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) {
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);
}
// if(!ObjectUtil.isEmpty(value)){
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;
}
item = (Item) var18.next();
ItemState itemState = (ItemState) itemStatus.get(item);
value = OpcUtl.getValue(item, itemState);
his = accessor_value.getValue(item.getId());
if (!ObjectUtil.equal(itemState.getQuality(), QualityTypeValue.OPC_QUALITY_GOOD) && his != null) {
log.warn("opc 值不健康 item: {}, 状态: {}", item.getId(), itemState.getQuality());
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);
}
} 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) (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);
}
log.info("信号{}变更从{}->{};信号快照：{}", new Object[]{item.getId(), his, value, sb});
ThreadUtl.sleep((long) ('\uea60' + random));
} else {
if (log.isWarnEnabled()) {
log.warn("{} 所有内容都为空, all_null:{} ,暂定{}ms", tag, all_null, 120000 + random);
}
ThreadUtl.sleep((long) (120000 + random));
}
//设置值
accessor_value.setValue(item.getId(), value);
++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) {
}
}
this.server = null;
if (!DeviceOpcSynchronizeAutoRun.isRun) {
log.warn("opc线程停止2。。。");
return;
}
String error_message = "设备信息同步异常";
if (!StrUtil.equals(this.message, error_message)) {
log.warn("", var30);
log.warn(error_message, var27);
}
try {
Thread.sleep((long) (OpcConfig.synchronized_exception_wait_second * 1000));
} catch (InterruptedException e) {
e.printStackTrace();
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;
}
}
}
}
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;
@ -204,4 +365,55 @@ 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) {
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});
}
}
}

7
hd/nladmin-system/src/main/java/org/nl/acs/opc/DeviceOpcSynchronizeAutoRun.java
View File

@ -17,6 +17,8 @@ import java.util.concurrent.Executors;
*/
@Component
public class DeviceOpcSynchronizeAutoRun extends AbstractAutoRunnable {
public static boolean isRun = false;
ExecutorService executorService = Executors.newCachedThreadPool();
@Autowired
private DeviceAppService deviceAppService;
@ -36,6 +38,7 @@ public class DeviceOpcSynchronizeAutoRun extends AbstractAutoRunnable {
@Override
public void autoRun() throws Exception {
{
isRun = true;
Map<String, OpcServerManageDto> servers = this.opcServerManageService.queryAllServerMap();
Map<String, List<List<OpcItemDto>>> pros;
@ -45,15 +48,12 @@ public class DeviceOpcSynchronizeAutoRun extends AbstractAutoRunnable {
}while (ObjectUtil.isEmpty(pros));
Set<String> keys = pros.keySet();
Iterator var4 = keys.iterator();
System.out.println("test:" + var4.hasNext());
//代码执行一次
while (var4.hasNext()) {
String key = (String) var4.next();
List<List<OpcItemDto>> list = (List) pros.get(key);
OpcServerManageDto opcServer = (OpcServerManageDto) servers.get(key);
Iterator var8 = list.iterator();
System.out.println("test2:" + var8.hasNext());
while (var8.hasNext()) {
List<OpcItemDto> groupProtols = (List) var8.next();
DeviceOpcProtocolRunable runable = new DeviceOpcProtocolRunable();
@ -75,6 +75,7 @@ public class DeviceOpcSynchronizeAutoRun extends AbstractAutoRunnable {
@Override
public void after() {
isRun = false;
this.executorService.shutdownNow();
this.executorService = Executors.newCachedThreadPool();
}

73
hd/nladmin-system/src/main/java/org/nl/acs/opc/ObjectUtl.java
View File

@ -0,0 +1,73 @@
package org.nl.acs.opc;
import org.nl.exception.WDKException;
import java.util.Arrays;
import java.util.Objects;
public class ObjectUtl {
private ObjectUtl() {
}
public static boolean isEquals(Object a, Object b) {
if (a == null && b == null) {
return true;
} else if (a != null && b != null) {
if (a.getClass().isArray()) {
if (a instanceof boolean[]) {
return Arrays.equals((boolean[]) ((boolean[]) a), (boolean[]) ((boolean[]) b));
} else if (a instanceof byte[]) {
return Arrays.equals((byte[]) ((byte[]) a), (byte[]) ((byte[]) b));
} else if (a instanceof int[]) {
return Arrays.equals((int[]) ((int[]) a), (int[]) ((int[]) b));
} else if (a instanceof long[]) {
return Arrays.equals((long[]) ((long[]) a), (long[]) ((long[]) b));
} else if (a instanceof double[]) {
return Arrays.equals((double[]) ((double[]) a), (double[]) ((double[]) b));
} else if (a instanceof short[]) {
return Arrays.equals((short[]) ((short[]) a), (short[]) ((short[]) b));
} else if (a instanceof char[]) {
return Arrays.equals((char[]) ((char[]) a), (char[]) ((char[]) b));
} else if (a instanceof float[]) {
return Arrays.equals((float[]) ((float[]) a), (float[]) ((float[]) b));
} else if (a instanceof Object[]) {
return Arrays.equals((Object[]) ((Object[]) a), (Object[]) ((Object[]) b));
} else {
throw new WDKException("未实现");
}
} else {
return Objects.equals(a, b);
}
} else {
return false;
}
}
public static boolean isTrue(Boolean boolean_) {
return boolean_ != null && isEquals(boolean_, true);
}
public static boolean isTrue(Boolean targetBoolean, boolean defaultBoolean) {
return targetBoolean == null ? defaultBoolean : targetBoolean;
}
public static boolean isFalse(Boolean boolean_) {
return boolean_ != null && isEquals(boolean_, false);
}
public static boolean isObject(Class<?> clazz) {
if (clazz == null) {
return false;
} else if (clazz.getClass().isArray()) {
return false;
} else {
return Object.class.isAssignableFrom(clazz);
}
}
}

2
hd/nladmin-system/src/main/java/org/nl/acs/opc/OpcConfig.java
View File

@ -10,4 +10,6 @@ public class OpcConfig {
public static String opc_server_default_group = "group";
public static String resource_code = "opc_sync";
public static String resource_name = "opc同步";
//OPC 数据同步是否采用回调机制实现。之前是线程定期全部读，效率低。
public static Boolean opc_item_read_using_callback = false;
}

21
hd/nladmin-system/src/main/java/org/nl/acs/opc/ThreadUtl.java
View File

@ -0,0 +1,21 @@
package org.nl.acs.opc;
/**
* @author ldjun
* @version 1.0
* @date 2023年02月01日 11:28
* @desc desc
*/
public class ThreadUtl {
private ThreadUtl() {
}
public static void sleep(long times) throws RuntimeException {
try {
Thread.sleep(times);
} catch (InterruptedException var3) {
throw new RuntimeException(var3);
}
}
}
Write Preview
Loading…
Cancel
Save