root/rongLiHongFeng
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

rev 读取plc线程更改

Browse Source
This commit is contained in:
gengby
2023-06-20 11:48:56 +08:00
parent d235df13aa
commit a3ef0faec3
8 changed files with 387 additions and 563 deletions
Download Patch File Download Diff File Expand all files Collapse all files

18
acs/nladmin-system/src/main/java/org/nl/acs/device/service/impl/ConveyorDevice.java
View File

@@ -44,15 +44,15 @@ public class ConveyorDevice {
@Override @Override
public void run() { public void run() {
System.out.println("test"); System.out.println("test");
String MODE = "RD1.RD1." + id + ".mode"; // String MODE = "RD1.RD1." + id + ".mode";
String code = OpcUtl.read(MODE); // String code = OpcUtl.read(MODE);
if ("2".equals(code)) { // if ("2".equals(code)) {
TaskService taskserver = new TaskServiceImpl(); // TaskService taskserver = new TaskServiceImpl();
TaskDto dto = new TaskDto(); // TaskDto dto = new TaskDto();
dto.setStart_point_code(id); // dto.setStart_point_code(id);
dto.setNext_point_code("1002"); // dto.setNext_point_code("1002");
taskserver.create(dto); // taskserver.create(dto);
} // }
} }
} }
} }

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

@@ -1,28 +1,36 @@
package org.nl.acs.opc; package org.nl.acs.opc;
import cn.hutool.core.date.DateUtil; import cn.hutool.core.date.DateUtil;
import cn.hutool.core.util.ObjectUtil;
import cn.hutool.core.util.StrUtil; import cn.hutool.core.util.StrUtil;
import lombok.extern.slf4j.Slf4j; import lombok.extern.slf4j.Slf4j;
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.wql.util.SpringContextHolder;
import org.openscada.opc.lib.da.ItemState; import org.openscada.opc.lib.da.*;
import org.openscada.opc.lib.da.Server;
import java.util.*; import java.util.*;
@Slf4j @Slf4j
public class DeviceOpcProtocolRunable implements Runnable { public class DeviceOpcProtocolRunable implements Runnable, DataCallback, ServerConnectionStateListener {
List<OpcItemDto> protocols; List<OpcItemDto> protocols;
OpcServerManageDto OpcServer; OpcServerManageDto OpcServer;
int error_num; int error_num;
String message; String message;
int maxResartNum;
private Server server;
private Group group;
boolean flag = false;
private int all_null;
private Map<String, OpcItemDto> itemSearchCache;
public DeviceOpcProtocolRunable() { public DeviceOpcProtocolRunable() {
this.error_num = 0; this.error_num = 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() {
@@ -41,32 +49,58 @@ public class DeviceOpcProtocolRunable implements Runnable {
this.OpcServer = opcServer; this.OpcServer = opcServer;
} }
OpcItemDto getItem(String item) {
Iterator var2 = this.protocols.iterator();
OpcItemDto dto; private OpcItemDto getItem(String item) {
do { OpcItemDto x = (OpcItemDto) this.itemSearchCache.get(item);
if (!var2.hasNext()) { if (x == null) {
return 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;
}
} }
}
dto = (OpcItemDto) var2.next(); return x;
} while (!StrUtil.equals(item, dto.getItem_code()));
return dto;
} }
@Override @Override
public void run() { public void run() {
while (true) { if (OpcConfig.opc_item_read_using_callback) {
try { this.runNew();
Server server = OpcServerUtl.getServerWithOutException(this.OpcServer.getOpc_host(), this.OpcServer.getCls_id(), this.OpcServer.getUser(), this.OpcServer.getPassword(), this.OpcServer.getDomain()); } else {
Group group = server.addGroup(); this.runOld();
List<String> itemsString = new ArrayList(); }
Iterator it = this.protocols.iterator(); }
while (it.hasNext()) {
OpcItemDto protocol = (OpcItemDto) it.next(); 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();
List<String> itemsString = new ArrayList();
Iterator var3 = this.protocols.iterator();
while (var3.hasNext()) {
OpcItemDto protocol = (OpcItemDto) var3.next();
String item = protocol.getItem_code(); String item = protocol.getItem_code();
itemsString.add(item); itemsString.add(item);
} }
@@ -74,47 +108,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);
itemsMap.put(string, item);
log.trace("添加成功 {}", string); log.trace("添加成功 {}", string);
} catch (Exception var29) { } catch (Exception var26) {
err_message.append(string + ":" + var29.getMessage()); 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 (true) {
long begin = System.currentTimeMillis(); 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])); Map<Item, ItemState> itemStatus = group.read(true, (Item[]) itemsMap.values().toArray(new Item[0]));
long end = System.currentTimeMillis(); long end = System.currentTimeMillis();
log.trace("{} 开始记时{}", tag, DateUtil.now());
long duration = end - begin; long duration = end - begin;
log.trace("{} 读取耗时：{}", tag, duration); if (log.isTraceEnabled()) {
log.trace("{} 读取耗时：{}", tag, duration);
}
if (duration > 1000L) { if (duration > 1000L) {
if (!time_out) { if (!time_out) {
log.warn("{} 读取超时 : {}", tag, duration); log.warn("{} 读取超时 : {}", tag, duration);
@@ -125,78 +169,198 @@ public class DeviceOpcProtocolRunable implements Runnable {
time_out = false; time_out = false;
} }
boolean valueAllNotNull = false;
Set<Item> items = itemStatus.keySet(); Set<Item> items = itemStatus.keySet();
Iterator var18 = items.iterator(); Iterator var18 = items.iterator();
while (true) { while (var18.hasNext()) {
Item item; Item item = (Item) var18.next();
//当前值 ItemState itemState = (ItemState) itemStatus.get(item);
Object value; Object value = OpcUtl.getValue(item, itemState);
//旧的值 if (value != null) {
Object his; valueAllNotNull = true;
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;
}
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());
}
} 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 + ";");
}
log.info("信号{}变更从{}->{};信号快照：{}", new Object[]{item.getId(), his, value, sb});
} }
//设置值
accessor_value.setValue(item.getId(), value);
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);
}
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) {
if (log.isWarnEnabled()) {
log.warn("{} 所有内容都为空, all_null:{} ,暂定{}s", tag, all_null, 5000 + random);
}
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);
}
ThreadUtl.sleep((long) ('\uea60' + random));
} else {
if (log.isWarnEnabled()) {
log.warn("{} 所有内容都为空, all_null:{} ,暂定{}ms", tag, all_null, 120000 + random);
}
ThreadUtl.sleep((long) (120000 + random));
}
++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) {
} }
} }
} catch (Exception var30) {
String error_message = "设备信息同步异常"; this.server = null;
if (!StrUtil.equals(this.message, error_message)) { if (!DeviceOpcSynchronizeAutoRun.isRun) {
log.warn("", var30); log.warn("opc线程停止2。。。");
return;
} }
try { String error_message = "设备信息同步异常";
Thread.sleep((long) (OpcConfig.synchronized_exception_wait_second * 1000)); if (!StrUtil.equals(this.message, error_message)) {
} catch (InterruptedException e) { log.warn(error_message, var27);
e.printStackTrace();
} }
ThreadUtl.sleep((long) (OpcConfig.synchronized_exception_wait_second * 1000));
++this.error_num; ++this.error_num;
if (this.error_num > 3 && !StrUtil.equals(this.message, error_message)) { if (this.error_num > 3 && !StrUtil.equals(this.message, error_message)) {
log.info("设备同步通信异常");
this.message = error_message; this.message = error_message;
} }
} }
} }
} }
private void logItemChanged(String itemId, UnifiedDataAccessor accessor_value, Object value, OpcItemDto itemDto) {
}
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)) {
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) { public static String formatDuring(long mss) {
long days = mss / 86400000L; long days = mss / 86400000L;
long hours = mss % 86400000L / 3600000L; long hours = mss % 86400000L / 3600000L;
@@ -204,4 +368,65 @@ public class DeviceOpcProtocolRunable implements Runnable {
long seconds = mss % 60000L / 1000L; long seconds = mss % 60000L / 1000L;
return days + " days " + hours + " hours " + minutes + " minutes " + seconds + " seconds "; return days + " days " + hours + " hours " + minutes + " minutes " + seconds + " seconds ";
} }
@Override
public void changed(Item item, ItemState itemState) {
}
// 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) {
// LuceneExecuteLogService lucene = SpringContextHolder.getBean(LuceneExecuteLogService.class);
// 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 (!itemDto.getItem_code().endsWith("heartbeat")) {
// lucene.deviceExecuteLog(new LuceneLogDto(itemDto.getOpc_server_code(), itemDto.getOpc_plc_code(), itemDto.getDevice_code(), itemDto.getItem_code().substring(itemDto.getItem_code().lastIndexOf(".")+1), String.valueOf(his), String.valueOf(value)));
// }
//
// } else {
// if (!itemDto.getItem_code().endsWith("heartbeat")) {
// lucene.deviceExecuteLog(new LuceneLogDto(itemDto.getOpc_server_code(), itemDto.getOpc_plc_code(), itemDto.getDevice_code(), itemDto.getItem_code().substring(itemDto.getItem_code().lastIndexOf(".")+1), String.valueOf(his), String.valueOf(value)));
// }
//
// }
// }
} }

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

@@ -1,276 +1,82 @@
package org.nl.acs.opc; package org.nl.acs.opc;
import cn.hutool.core.thread.NamedThreadFactory;
import cn.hutool.core.util.ObjectUtil; import cn.hutool.core.util.ObjectUtil;
import lombok.extern.slf4j.Slf4j;
import org.nl.acs.auto.run.AbstractAutoRunnable; import org.nl.acs.auto.run.AbstractAutoRunnable;
import org.nl.acs.udw.UnifiedDataAccessor;
import org.nl.acs.udw.UnifiedDataAccessorFactory;
import org.nl.acs.udw.UnifiedDataAppService;
import org.nl.modules.wql.util.SpringContextHolder;
import org.openscada.opc.lib.da.Group;
import org.openscada.opc.lib.da.Item;
import org.openscada.opc.lib.da.ItemState;
import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component; import org.springframework.stereotype.Component;
import java.util.*; import java.util.Iterator;
import java.util.concurrent.*; import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
/** /**
* OPC设备同步启动 * OPC设备同步启动
*/ */
@Component @Component
@Slf4j
public class DeviceOpcSynchronizeAutoRun extends AbstractAutoRunnable { public class DeviceOpcSynchronizeAutoRun extends AbstractAutoRunnable {
static boolean isRun = true;
public static boolean isRun = false;
ExecutorService executorService = Executors.newCachedThreadPool();
@Autowired @Autowired
private DeviceAppService deviceAppService; private DeviceAppService deviceAppService;
@Autowired @Autowired
private OpcServerManageService opcServerManageService; private OpcServerManageService opcServerManageService;
// @Autowired
// LuceneExecuteLogService lucene;
static ExecutorService executorService;
public static Map<String, OpcServerManageDto> opcServersConfig;
public static Map<String, OpcItemDto> itemCodeOpcItemDtoMapping = new ConcurrentHashMap();
static boolean canRefreshOpcEntity = true;
private long lastRefreshOpcEntityTime;
static UnifiedDataAccessor udw;
private static Map<String, Boolean> canReadOpcValues;
private static volatile Map<String, OpcEntity> opcCodeOpcEntityMapping;
public DeviceOpcSynchronizeAutoRun() {
this.lastRefreshOpcEntityTime = 0L;
}
@Override
public String getCode() { public String getCode() {
return DeviceOpcSynchronizeAutoRun.class.getSimpleName(); return DeviceOpcSynchronizeAutoRun.class.getSimpleName();
} }
@Override
public String getName() { public String getName() {
return "opc设备同步器"; return "opc设备同步器";
} }
static Group getGroup(String opcCode) throws Exception { @Override
OpcServerService opcServerService = SpringContextHolder.getBean(OpcServerService.class); public void autoRun() throws Exception {
return opcServerService.getServer(opcCode); {
} isRun = true;
static void submitTimeLimitTask(Runnable runnable, String opcCode) {
CompletableFuture<Void> future = CompletableFuture.runAsync(runnable, executorService);
// try {
// future.get(10L, TimeUnit.SECONDS);
// } catch (InterruptedException var9) {
// Thread.currentThread().interrupt();
// } catch (ExecutionException var10) {
// var10.printStackTrace();
// } catch (TimeoutException var11) {
// itemCodeOpcItemDtoMapping.keySet().forEach((key) -> {
// udw.setValue(key, (Object) null);
// });
// canReadOpcValues = new ConcurrentHashMap<>();
// System.out.println("opc设备同步器 任务执行超时，取消任务...");
// future.cancel(true);
// } finally {
// canRefreshOpcEntity = true;
// if (opcCode != null) {
// canReadOpcValues.put(opcCode, true);
// }
//
// }
}
private ExecutorService createThreadPool() {
ThreadPoolExecutor executor = new ThreadPoolExecutor(16, 16, 10L, TimeUnit.SECONDS, new LinkedBlockingQueue<>(), new NamedThreadFactory("opc-sync", false));
executor.allowCoreThreadTimeOut(true);
return executor;
}
public void autoRun() {
OpcStartTag.is_run = true;
opcServersConfig = this.opcServerManageService.queryAllServerMap();
executorService = this.createThreadPool();
opcCodeOpcEntityMapping = new ConcurrentHashMap();
itemCodeOpcItemDtoMapping.keySet().forEach((key) -> {
udw.setValue(key, (Object) null);
});
canRefreshOpcEntity = true;
canReadOpcValues.clear();
while (true) {
this.refreshOpcEntity();
Iterator var1 = opcServersConfig.keySet().iterator();
while (var1.hasNext()) {
String opcCode = (String) var1.next();
submitTimeLimitTask(() -> {
boolean in = false;
try {
if (canReadOpcValues.computeIfAbsent(opcCode, (key) -> true)) {
in = true;
canReadOpcValues.put(opcCode, false);
this.readOpcValues(opcCode);
}
} catch (Exception var3) {
var3.printStackTrace();
} finally {
canRefreshOpcEntity = true;
if (opcCode != null && in) {
canReadOpcValues.put(opcCode, true);
}
}
}, opcCode);
}
ThreadUtl.sleep((long) OpcConfig.synchronized_millisecond);
}
}
private void readOpcValues(String opcCode) throws Exception {
synchronized (opcCode.intern()) {
OpcEntity opcEntity = (OpcEntity) opcCodeOpcEntityMapping.get(opcCode);
if (opcEntity != null) {
if (opcEntity.getItems().size() != 0) {
long begin = System.currentTimeMillis();
if (log.isTraceEnabled()) {
log.trace("opc {} 开始计时{}", opcCode, begin);
}
Map<Item, ItemState> itemStatus;
try {
itemStatus = opcEntity.readAll();
} catch (Exception var15) {
itemStatus = opcEntity.readDividually();
}
long end = System.currentTimeMillis();
long duration = end - begin;
if (log.isTraceEnabled()) {
log.trace("opc {} 读取耗时：{}", opcCode, duration);
}
if (duration > 1000L) {
log.warn("opc {} 读取超时 : {}", opcCode, duration);
}
// boolean allNull = itemStatus.entrySet().stream().map((map) -> {
// return OpcUtl.getValue((Item)map.getKey(), (ItemState)map.getValue());
// }).allMatch(Objects::isNull);
// if (allNull) {
// opcEntity.getItems().clear();
// }
UnifiedDataAccessor udw = opcEntity.getUdw();
Set<Item> items = itemStatus.keySet();
Iterator var18 = items.iterator();
while (var18.hasNext()) {
Item item = (Item) var18.next();
ItemState itemState = (ItemState) itemStatus.get(item);
Object nowValue = OpcUtl.getValue(item, itemState);
String itemId = item.getId();
Object historyValue = udw.getValue(itemId);
if (!ObjectUtl.isEquals(itemState.getQuality(), QualityTypeValue.OPC_QUALITY_GOOD) && historyValue != null) {
log.warn("opc 值不健康 item: {}, 状态: {}", itemId, itemState.getQuality());
}
if (!UnifiedDataAppService.isEquals(nowValue, historyValue)) {
OpcItemDto itemDto = (OpcItemDto) itemCodeOpcItemDtoMapping.get(itemId);
if (true) {
this.logItemChanged(itemId, udw, nowValue, itemDto);
}
udw.setValue(itemId, nowValue);
}
}
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));
Set<String> keys = pros.keySet();
Iterator var4 = keys.iterator();
//代码执行一次
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();
while (var8.hasNext()) {
List<OpcItemDto> groupProtols = (List) var8.next();
DeviceOpcProtocolRunable runable = new DeviceOpcProtocolRunable();
runable.setProtocols(groupProtols);
runable.setOpcServer(opcServer);
this.executorService.submit(runable);
} }
} }
}
}
private void refreshOpcEntity() { // 同步无光电设备信号
if (canRefreshOpcEntity) { //Map<String, List<List<OpcItemDto>>> pros1 = this.deviceAppService.findAllFormatProtocolFromDriver();
canRefreshOpcEntity = false; //List<DeviceDriver> opcDrivers = this.deviceAppService.findDeviceDriver(DeviceDriver.class);
long now = System.currentTimeMillis();
if (now - this.lastRefreshOpcEntityTime >= 20000L) {
this.lastRefreshOpcEntityTime = now;
submitTimeLimitTask(() -> {
try {
Map<String, List<List<OpcItemDto>>> protocol = this.deviceAppService.findAllFormatProtocolFromDriver();
Iterator var2 = protocol.entrySet().iterator();
while (var2.hasNext()) { while (true) {
Map.Entry<String, List<List<OpcItemDto>>> stringListEntry = (Map.Entry) var2.next(); Thread.sleep(3000L);
String opcCode = (String) stringListEntry.getKey();
List<List<OpcItemDto>> opcItemDtos = (List) stringListEntry.getValue();
((OpcEntity) opcCodeOpcEntityMapping.computeIfAbsent(opcCode, OpcEntity::new)).reload(opcItemDtos);
}
} catch (Exception var6) {
var6.printStackTrace();
} finally {
canRefreshOpcEntity = true;
}
}, (String) null);
} }
} }
} }
private void logMessage(String errorMessage) { @Override
try {
// issueLogger.setResource(OpcConfig.resource_code, OpcConfig.resource_name).setError(StringUtl.getString(100), "设备同步通信异常").log(errorMessage, new Object[0]);
// businessLogger.setResource(OpcConfig.resource_code, OpcConfig.resource_name).setError(StringUtl.getString(100), "设备同步通信异常").log(errorMessage, new Object[0]);
} catch (Exception var5) {
var5.printStackTrace();
}
}
public void after() { public void after() {
OpcStartTag.is_run = false; isRun = false;
opcCodeOpcEntityMapping.values().forEach((opcEntity) -> { this.executorService.shutdownNow();
opcEntity.cleanUdwCache(); this.executorService = Executors.newCachedThreadPool();
OpcServerService opcServerService = SpringContextHolder.getBean(OpcServerService.class);
opcServerService.cleanGroups(opcEntity.getOpcCode());
});
opcCodeOpcEntityMapping = new ConcurrentHashMap();
itemCodeOpcItemDtoMapping = new ConcurrentHashMap();
executorService.shutdownNow();
}
private void logItemChanged(String itemCode, UnifiedDataAccessor udw, Object value, OpcItemDto itemDto) {
Object his = udw.getValue(itemCode);
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 = udw.getValue(relate);
sb.append("key:").append(relate).append("value:").append(obj).append(";");
}
// if (!itemCode.endsWith("heartbeat") && !itemCode.endsWith("time")) {
// log.warn("{} 信号 {} 发生变更 {} -> {} 信号快照 {}", itemDto.getDevice_code(), itemCode, his, value, sb);
// lucene.deviceExecuteLog(new LuceneLogDto(itemDto.getOpc_server_code(), itemDto.getOpc_plc_code(), itemDto.getDevice_code(), itemDto.getItem_code().substring(itemDto.getItem_code().lastIndexOf(".") + 1), String.valueOf(his), String.valueOf(value)));
// }
} else {
// if (!itemCode.endsWith("heartbeat") && !itemCode.endsWith("time")) {
// log.warn("{} 信号 {} 发生变更 {} -> {}", itemDto.getDevice_code(), itemCode, his, value);
// lucene.deviceExecuteLog(new LuceneLogDto(itemDto.getOpc_server_code(), itemDto.getOpc_plc_code(), itemDto.getDevice_code(), itemDto.getItem_code().substring(itemDto.getItem_code().lastIndexOf(".") + 1), String.valueOf(his), String.valueOf(value)));
// }
}
}
static {
udw = UnifiedDataAccessorFactory.getAccessor(OpcConfig.udw_opc_value_key);
canReadOpcValues = new ConcurrentHashMap();
opcCodeOpcEntityMapping = new ConcurrentHashMap();
} }
} }

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

@@ -4,10 +4,13 @@ public class OpcConfig {
public static Boolean auto_start_opc = Boolean.valueOf(true); public static Boolean auto_start_opc = Boolean.valueOf(true);
public static String udw_opc_value_key = "opc_value"; public static String udw_opc_value_key = "opc_value";
public static Integer synchronized_millisecond = Integer.valueOf(100); public static Integer synchronized_millisecond = Integer.valueOf(100);
public static Integer synchronized_exception_wait_second = Integer.valueOf(10); public static Integer synchronized_exception_wait_second = Integer.valueOf(3);
public static Integer retry_times = Integer.valueOf(3); public static Integer retry_times = Integer.valueOf(3);
public static String sync_issue_type_code = "device_opc_sync"; public static String sync_issue_type_code = "device_opc_sync";
public static String opc_server_default_group = "group"; public static String opc_server_default_group = "group";
public static String resource_code = "opc_sync"; public static String resource_code = "opc_sync";
public static String resource_name = "opc同步"; public static String resource_name = "opc同步";
//OPC 数据同步是否采用回调机制实现。之前是线程定期全部读，效率低。
public static Boolean opc_item_read_using_callback = false;
} }

176
acs/nladmin-system/src/main/java/org/nl/acs/opc/OpcEntity.java
View File

@@ -1,176 +0,0 @@
package org.nl.acs.opc;
import org.nl.acs.udw.UnifiedDataAccessor;
import org.nl.acs.udw.UnifiedDataAccessorFactory;
import org.nl.modules.wql.util.SpringContextHolder;
import org.openscada.opc.lib.da.Group;
import org.openscada.opc.lib.da.Item;
import org.openscada.opc.lib.da.ItemState;
import java.util.*;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.stream.Collectors;
public class OpcEntity {
OpcServerService opcServerService = SpringContextHolder.getBean(OpcServerService.class);
private final UnifiedDataAccessor udw;
private Map<String, List<Item>> items;
private Map<String, List<String>> someFailDevices;
private String opcCode;
public OpcEntity(String opcCode) {
this.udw = UnifiedDataAccessorFactory.getAccessor(OpcConfig.udw_opc_value_key);
this.items = new ConcurrentHashMap();
this.someFailDevices = new ConcurrentHashMap();
this.opcCode = opcCode;
}
public void reload(List<List<OpcItemDto>> opcItemDtos) {
Map<String, List<String>> itemCodes = new ConcurrentHashMap();
(opcItemDtos.stream().flatMap(Collection::stream).collect(Collectors.groupingBy(OpcItemDto::getDevice_code))).forEach((deviceCodes, opcItemDtoList) -> {
itemCodes.put(deviceCodes, opcItemDtoList.stream().map(OpcItemDto::getItem_code).collect(Collectors.toList()));
});
DeviceOpcSynchronizeAutoRun.itemCodeOpcItemDtoMapping.putAll((Map)opcItemDtos.stream().flatMap(Collection::stream).collect(Collectors.toMap(OpcItemDto::getItem_code, (obj) -> {
return obj;
},(k, v) -> k)));
if (this.items.size() == 0) {
itemCodes.values().stream().flatMap(Collection::stream).forEach((key) -> {
this.udw.setValue(key, (Object)null);
});
this.addItemsIntoGroup(itemCodes);
} else {
if (this.someFailDevices.size() > 0) {
this.reAddDevices();
}
}
}
private void reAddDevices() {
Map<String, List<Item>> addItems = new ConcurrentHashMap();
StringBuilder err_message = new StringBuilder();
this.someFailDevices.forEach((deviceCode, itemCodesList) -> {
itemCodesList.forEach((itemCode) -> {
try {
Group group = DeviceOpcSynchronizeAutoRun.getGroup(this.opcCode);
((List)addItems.computeIfAbsent(deviceCode, (key) -> {
return new ArrayList();
})).add(group.addItem(itemCode));
} catch (Exception var6) {
err_message.append(itemCode).append(" 添加失败; ");
}
});
List<Item> deviceItems = (List)addItems.get(deviceCode);
if (deviceItems != null && deviceItems.size() == itemCodesList.size()) {
this.someFailDevices.remove(deviceCode);
} else if (itemCodesList.size() == 0) {
addItems.remove(deviceCode);
} else {
assert deviceItems != null;
((List)this.someFailDevices.get(deviceCode)).removeAll(deviceItems);
}
synchronized(this.opcCode.intern()) {
this.items.putAll(addItems);
}
if (err_message.length() > 0) {
String errMsg = err_message.toString();
//this.log.warn("{}:{}", com.wxzd.wcs.opc.OpcConfig.resource_code, errMsg);
}
});
}
private void addItemsIntoGroup(Map<String, List<String>> itemCodes) {
try {
Group group = DeviceOpcSynchronizeAutoRun.getGroup(this.opcCode);
StringBuilder err_message = new StringBuilder();
Map<String, List<Item>> items = new ConcurrentHashMap();
itemCodes.forEach((deviceCode, itemCodesList) -> {
itemCodesList.forEach((itemCode) -> {
try {
((List)items.computeIfAbsent(deviceCode, (key) -> {
return new ArrayList();
})).add(group.addItem(itemCode));
} catch (Exception var7) {
((List)this.someFailDevices.computeIfAbsent(deviceCode, (key) -> {
return new ArrayList();
})).add(itemCode);
this.udw.setValue(itemCode, (Object)null);
err_message.append(itemCode).append(" 添加失败; ");
}
});
List<Item> deviceItems = (List)items.get(deviceCode);
if (deviceItems != null && deviceItems.size() != itemCodesList.size()) {
items.remove(deviceCode);
this.someFailDevices.put(deviceCode, itemCodesList);
}
});
synchronized(this.opcCode.intern()) {
this.items = items;
}
if (err_message.length() > 0) {
String errMsg = err_message.toString();
// this.log.warn("{}:{}", OpcConfig.resource_code, errMsg);
}
} catch (Exception var8) {
var8.printStackTrace();
}
}
public void cleanUdwCache() {
this.items.values().stream().flatMap(Collection::stream).map(Item::getId).forEach((key) -> {
this.udw.setValue(key, (Object)null);
});
}
public Map<Item, ItemState> readAll() throws Exception {
return opcServerService.getServer(this.opcCode).read(true, (Item[])this.items.values().stream().flatMap(Collection::stream).toArray((x$0) -> {
return new Item[x$0];
}));
}
public Map<Item, ItemState> readDividually() {
Map<Item, ItemState> result = new HashMap();
CompletableFuture[] futures = (CompletableFuture[])this.items.entrySet().stream().map((entry) -> {
return CompletableFuture.runAsync(() -> {
try {
Group group = DeviceOpcSynchronizeAutoRun.getGroup(this.opcCode);
result.putAll(group.read(true, (Item[])((List)entry.getValue()).toArray(new Item[0])));
} catch (Exception var5) {
String deviceCode = (String)entry.getKey();
// to do
// this.someFailDevices.put(deviceCode, ((List)entry.getValue()).stream().map(Item::getId).collect(Collectors.toList()));
this.items.remove(deviceCode);
}
}, DeviceOpcSynchronizeAutoRun.executorService);
}).toArray((x$0) -> {
return new CompletableFuture[x$0];
});
CompletableFuture.allOf(futures).join();
return result;
}
public UnifiedDataAccessor getUdw() {
return this.udw;
}
public Map<String, List<Item>> getItems() {
return this.items;
}
public String getOpcCode() {
return this.opcCode;
}
}

23
acs/nladmin-system/src/main/java/org/nl/acs/opc/OpcServerService.java
View File

@@ -1,23 +0,0 @@
package org.nl.acs.opc;
import org.nl.acs.device_driver.driver.ItemValue;
import org.openscada.opc.lib.da.Group;
/**
* @author ldjun
* @version 1.0
* @date 2023年02月01日 11:26
* @desc desc
*/
public interface OpcServerService {
void reload();
Group getServer(String var1) throws Exception;
void writeInteger(String var1, ItemValue... var2);
void clearServer(String var1);
void cleanGroups(String var1);
}

31
acs/nladmin-system/src/main/java/org/nl/acs/opc/OpcUtl.java
View File

@@ -40,13 +40,13 @@ public class OpcUtl {
Map<Item, Integer> e=null; Map<Item, Integer> e=null;
try{ try{
e=group.write(requests); e=group.write(requests);
group.write(requests); //group.write(requests);
}catch (Exception e1){ }catch (Exception e1){
try{ try{
e= group.write(requests); e= group.write(requests);
}catch (Exception e2){ }catch (Exception e2){
e= group.write(requests); e= group.write(requests);
} }
} }
boolean is_success = true; boolean is_success = true;
@@ -78,7 +78,7 @@ public class OpcUtl {
} catch (JIException arg7) { } catch (JIException arg7) {
log.info("下发信号失败："+arg7.getMessage()); log.info("下发信号失败："+arg7.getMessage());
System.out.println("下发信号失败原因："+arg7.getMessage()); System.out.println("下发信号失败原因："+arg7.getMessage());
throw new WDKException(arg7); throw new WDKException(arg7);
} }
} }
@@ -226,11 +226,12 @@ public class OpcUtl {
return server; return server;
} catch (Exception e) { } catch (Exception e) {
errNum++; errNum++;
// e.printStackTrace();
System.out.println("server error:"+e.getMessage());
throw new WDKException(e.getMessage()); throw new WDKException(e.getMessage());
}finally{ }finally{
System.out.println("successNum:"+successNum); System.out.println("successNum:"+successNum);
System.out.println("errNum:"+errNum); System.out.println("errNum:"+errNum);
System.out.println(11);
} }
} }
@@ -292,18 +293,6 @@ public class OpcUtl {
return value; return value;
} }
public static String read(String item) throws Exception {
System.out.println(item);
Server server = getServer("192.168.81.251", "7bc0cc8e-482c-47ca-abdc-0fe7f9c6e729", "administrator", "Huawei@123", "");
// String byteItemString = "RD1.RD1.1001.mode";
Group group = server.addGroup();
Item byteItem = group.addItem(item);
ItemState itemState = null;
JIVariant value = null;
itemState = byteItem.read(true);
value = itemState.getValue();
String data = OpcUtl.getValue(byteItem, itemState) + "";
return data;
}
} }

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

@@ -128,7 +128,7 @@ file:
avatarMaxSize: 5 avatarMaxSize: 5
logging: logging:
file: file:
path: C:\log\acs path: D:\ACS_LOG
config: classpath:logback-spring.xml config: classpath:logback-spring.xml
# Sa-Token配置 # Sa-Token配置