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

fix opc线程

master
汪菘 2 years ago
parent
1e961e0e3d
commit
6125320428
4 changed files with 307 additions and 120 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. 27
      hd/nladmin-system/src/main/java/org/nl/acs/opc/DeviceOpcSynchronizeAutoRun.java
  3. 1
      hd/nladmin-system/src/main/java/org/nl/acs/opc/OpcConfig.java
  4. 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

@ -6,24 +6,31 @@ 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.*;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
@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;
}
@ -43,36 +50,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 {
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.OpcServer.getOpc_host());
Group group = server.addGroup(this.OpcServer.getOpc_code());
if(ObjectUtil.isEmpty(group)){
log.info("group is null ");
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);
}
@ -80,58 +109,60 @@ 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();
Map<Item, ItemState> itemStatus = null;
try {
itemStatus = group.read(true, (Item[]) itemsMap.values().toArray(new Item[0]));
} catch (Exception e){
System.out.println("数据同步异常:"+ this.getOpcServer().getOpc_code());
log.trace("数据同步异常:{}", this.getOpcServer().getOpc_code());
//e.printStackTrace();
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);
System.out.println("线程："+tag + " 读取耗时:"+ duration);
if (log.isTraceEnabled()) {
log.trace("{} 读取耗时：{}", tag, duration);
}
if (duration > 1000L) {
if (!time_out) {
log.warn(" {} 读取超时 : {}", tag, duration);
log.warn("{} 读取超时 : {}", tag, duration);
}
time_out = true;
@ -139,99 +170,183 @@ public class DeviceOpcProtocolRunable implements Runnable {
time_out = false;
}
if(ObjectUtil.isEmpty(itemStatus)) {
System.out.println( tag + " :itemStatus is null");
log.warn(" {} 读取异常 : {} itemStatus is null", tag);
}
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;
}
item = (Item) var18.next();
ItemState itemState = (ItemState) itemStatus.get(item);
value = OpcUtl.getValue(item, itemState);
his = accessor_value.getValue(item.getId());
if (!ObjectUtl.isEquals(itemState.getQuality(), QualityTypeValue.OPC_QUALITY_GOOD) && his != null) {
log.warn("opc 值不健康 item: {}, 状态: {},当前读取值：{}, 系统内存值{} ", item.getId(), itemState.getQuality(), value, his);
}
} while (ObjectUtil.equal(value, his));//如果两次的值相等，不走下面的代码
OpcItemDto itemDto = this.getItem(item.getId());
//默认记录日志
if (true) {
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 + ";");
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);
}
log.warn("信号{}变更从{}->{};信号快照：{}", new Object[]{item.getId(), his, value, sb});
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;
}
if (!valueAllNotNull) {
int random = (new Random()).nextInt(10) + 1;
random *= 1000;
if (this.all_null < 3) {
log.warn("{} 所有内容都为空, all_null:{} ,暂定{}s", tag, all_null,5000 + random);
ThreadUtl.sleep((long) (5000 + random));
} else if (this.all_null < 6) {
log.warn("{} 所有内容都为空, all_null:{} ,暂定{}s", tag, all_null,30000 + random);
ThreadUtl.sleep((long) (1000 + random));
log.warn("重新创建server");
break start;
} else if (this.all_null < 12) {
log.warn("{} 所有内容都为空, all_null:{} ,暂定{}ms", tag, all_null, '\uea60' + random);
ThreadUtl.sleep((long) ('\uea60' + random));
} else {
log.warn("{} 所有内容都为空, all_null:{} ,暂定{}ms", tag, all_null, 120000 + random);
ThreadUtl.sleep((long) (120000 + random));
}
// accessor_value.setValueWithPersistence(item.getId(),accessor_value.getValue(item.getId()));
// accessor_value.getHistoryUnifiedData(item.getId());
//设置值
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) {
log.warn("{} : server disconnect", var25);
}
}
this.server = null;
if (!DeviceOpcSynchronizeAutoRun.isRun) {
log.warn("opc线程停止2。。。");
return;
}
String error_message = "设备信息同步异常,"+var30;
String error_message = "设备信息同步异常";
if (!StrUtil.equals(this.message, error_message)) {
log.warn("", var30);
log.warn(error_message, var27);
}
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;
}
try {
Thread.sleep((long) (OpcConfig.synchronized_exception_wait_second * 1000));
} catch (InterruptedException e) {
e.printStackTrace();
log.warn("OPC 读取线程读取异常{} :", e);
// Thread.currentThread().interrupt();不会真正停止线程
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;
@ -239,4 +354,49 @@ 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);
log.trace("Item {} new value: {}, Timestamp: {}", new Object[]{itemId, itemState.getValue(), itemState.getTimestamp().getTime()});
OpcItemDto itemDto = this.getItem(itemId);
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 + ";");
}
if(!itemId.endsWith("heartbeat") || !itemId.endsWith("time") ) {
log.warn("设备：{}信号{}变更从{}->{};信号快照：{}", new Object[]{itemDto.getDevice_code(), itemId, his, value, sb});
}
} else {
if(!itemId.endsWith("heartbeat") || !itemId.endsWith("time") ) {
log.warn("设备：{}信号{}变更从{}->{};信号快照：{}", new Object[]{itemDto.getDevice_code(), itemId, his, value});
}
}
}
}

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

@ -17,7 +17,9 @@ import java.util.concurrent.Executors;
*/
@Component
public class DeviceOpcSynchronizeAutoRun extends AbstractAutoRunnable {
ExecutorService executorService = Executors.newCachedThreadPool();
public static boolean isRun = false;
ExecutorService executorService;
@Autowired
private DeviceAppService deviceAppService;
@Autowired
@ -36,24 +38,22 @@ public class DeviceOpcSynchronizeAutoRun extends AbstractAutoRunnable {
@Override
public void autoRun() throws Exception {
{
isRun = true;
this.executorService = Executors.newCachedThreadPool();
Map<String, OpcServerManageDto> servers = this.opcServerManageService.queryAllServerMap();
Map<String, List<List<OpcItemDto>>> pros;
do{
Thread.sleep(1000L);
pros = this.deviceAppService.findAllFormatProtocolFromDriver();
}while (ObjectUtil.isEmpty(pros));
do{
Thread.sleep(1000L);
pros = this.deviceAppService.findAllFormatProtocolFromDriver();
}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();
@ -67,14 +67,19 @@ public class DeviceOpcSynchronizeAutoRun extends AbstractAutoRunnable {
//Map<String, List<List<OpcItemDto>>> pros1 = this.deviceAppService.findAllFormatProtocolFromDriver();
//List<DeviceDriver> opcDrivers = this.deviceAppService.findDeviceDriver(DeviceDriver.class);
while (true) {
Thread.sleep(3000L);
while(true) {
try {
Thread.sleep(3600000L);
} catch (InterruptedException var11) {
return;
}
}
}
}
@Override
public void after() {
isRun = false;
this.executorService.shutdownNow();
this.executorService = Executors.newCachedThreadPool();
}

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

@ -10,4 +10,5 @@ public class OpcConfig {
public static String opc_server_default_group = "group";
public static String resource_code = "opc_sync";
public static String resource_name = "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