25 changed files with 1492 additions and 255 deletions
@ -1,81 +1,305 @@ |
|||
//
|
|||
// Source code recreated from a .class file by IntelliJ IDEA
|
|||
// (powered by FernFlower decompiler)
|
|||
//
|
|||
|
|||
package org.nl.acs.opc; |
|||
|
|||
import cn.hutool.core.util.ObjectUtil; |
|||
import java.util.*; |
|||
import java.util.Map.Entry; |
|||
import java.util.concurrent.CompletableFuture; |
|||
import java.util.concurrent.ConcurrentHashMap; |
|||
import java.util.concurrent.ExecutionException; |
|||
import java.util.concurrent.ExecutorService; |
|||
import java.util.concurrent.LinkedBlockingQueue; |
|||
import java.util.concurrent.ThreadPoolExecutor; |
|||
import java.util.concurrent.TimeUnit; |
|||
import java.util.concurrent.TimeoutException; |
|||
|
|||
import lombok.extern.slf4j.Slf4j; |
|||
import org.apache.lucene.util.NamedThreadFactory; |
|||
import org.nl.acs.auto.run.AbstractAutoRunnable; |
|||
import org.nl.acs.task.service.TaskService; |
|||
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.stereotype.Component; |
|||
|
|||
import java.util.Iterator; |
|||
import java.util.List; |
|||
import java.util.Map; |
|||
import java.util.Set; |
|||
import java.util.concurrent.ExecutorService; |
|||
import java.util.concurrent.Executors; |
|||
|
|||
/** |
|||
* OPC设备同步启动 |
|||
*/ |
|||
@Component |
|||
@Slf4j |
|||
public class DeviceOpcSynchronizeAutoRun extends AbstractAutoRunnable { |
|||
ExecutorService executorService = Executors.newCachedThreadPool(); |
|||
static boolean isRun = true; |
|||
@Autowired |
|||
private DeviceAppService deviceAppService; |
|||
@Autowired |
|||
private OpcServerManageService opcServerManageService; |
|||
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() { |
|||
return DeviceOpcSynchronizeAutoRun.class.getSimpleName(); |
|||
} |
|||
|
|||
@Override |
|||
public String getName() { |
|||
return "opc设备同步器"; |
|||
} |
|||
|
|||
@Override |
|||
public void autoRun() throws Exception { |
|||
{ |
|||
|
|||
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(); |
|||
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(); |
|||
runable.setProtocols(groupProtols); |
|||
runable.setOpcServer(opcServer); |
|||
this.executorService.submit(runable); |
|||
static Group getGroup(String opcCode) throws Exception { |
|||
OpcServerService opcServerService = SpringContextHolder.getBean(OpcServerService.class); |
|||
return opcServerService.getServer(opcCode); |
|||
} |
|||
|
|||
static void submitTimeLimitTask(Runnable runnable, String opcCode) { |
|||
CompletableFuture 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(10, 30, 60L, TimeUnit.SECONDS, new LinkedBlockingQueue(), new NamedThreadFactory("opc-sync")); |
|||
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(); |
|||
Boolean canRead = (Boolean)canReadOpcValues.computeIfAbsent(opcCode, (key) -> { |
|||
return true; |
|||
}); |
|||
if (canRead) { |
|||
canReadOpcValues.put(opcCode, false); |
|||
submitTimeLimitTask(() -> { |
|||
try { |
|||
this.readOpcValues(opcCode); |
|||
} catch (Exception var3) { |
|||
var3.printStackTrace(); |
|||
} |
|||
|
|||
}, opcCode); |
|||
} |
|||
} |
|||
|
|||
// 同步无光电设备信号
|
|||
//Map<String, List<List<OpcItemDto>>> pros1 = this.deviceAppService.findAllFormatProtocolFromDriver();
|
|||
//List<DeviceDriver> opcDrivers = this.deviceAppService.findDeviceDriver(DeviceDriver.class);
|
|||
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); |
|||
} |
|||
|
|||
new HashMap(); |
|||
|
|||
Map 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();
|
|||
// }
|
|||
|
|||
while (true) { |
|||
Thread.sleep(3000L); |
|||
|
|||
|
|||
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); |
|||
} |
|||
|
|||
} |
|||
|
|||
// itemStatus.forEach((Item item, ItemState itemState) -> {
|
|||
// Object nowValue = null;
|
|||
// try {
|
|||
// nowValue = OpcUtl.getValue((Item)item, (ItemState)itemState);
|
|||
// } catch (Exception e) {
|
|||
// e.printStackTrace();
|
|||
// }
|
|||
// 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 (Boolean.TRUE.equals(itemDto.getNeed_log())) {
|
|||
// this.logItemChanged(itemId, udw, nowValue, itemDto);
|
|||
// }
|
|||
//
|
|||
// udw.setValue(itemId, nowValue);
|
|||
// }
|
|||
//
|
|||
// });
|
|||
|
|||
} |
|||
} |
|||
} |
|||
} |
|||
|
|||
@Override |
|||
private void refreshOpcEntity() { |
|||
if (canRefreshOpcEntity) { |
|||
canRefreshOpcEntity = false; |
|||
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()) { |
|||
Entry<String, List<List<OpcItemDto>>> stringListEntry = (Entry)var2.next(); |
|||
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(); |
|||
} |
|||
|
|||
}, (String)null); |
|||
} |
|||
} |
|||
} |
|||
|
|||
private void logMessage(String errorMessage) { |
|||
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() { |
|||
this.executorService.shutdownNow(); |
|||
this.executorService = Executors.newCachedThreadPool(); |
|||
OpcStartTag.is_run = false; |
|||
opcCodeOpcEntityMapping.values().forEach((opcEntity) -> { |
|||
opcEntity.cleanUdwCache(); |
|||
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(";"); |
|||
} |
|||
log.warn("设备：{}信号{}变更从{}->{};信号快照：{}", new Object[]{itemDto.getDevice_code(), itemCode, his, value, sb}); |
|||
System.out.println("设备："+itemDto.getDevice_code() + "信号："+ itemCode + itemCode + "变更从：" + his + "->" +value); |
|||
} else { |
|||
log.warn("设备：{}信号{}变更从{}->{};信号快照：{}", new Object[]{itemDto.getDevice_code(), itemCode, his, value}); |
|||
System.out.println("设备："+itemDto.getDevice_code() + "信号："+ itemCode + itemCode + "变更从：" + his + "->" +value); |
|||
} |
|||
|
|||
} |
|||
|
|||
static { |
|||
udw = UnifiedDataAccessorFactory.getAccessor(OpcConfig.udw_opc_value_key); |
|||
canReadOpcValues = new ConcurrentHashMap(); |
|||
opcCodeOpcEntityMapping = new ConcurrentHashMap(); |
|||
} |
|||
} |
|||
|
|||
@ -0,0 +1,30 @@ |
|||
package org.nl.acs.opc; |
|||
|
|||
public class ItemValue { |
|||
private String item_code; |
|||
private Object item_value; |
|||
|
|||
public ItemValue() { |
|||
} |
|||
|
|||
public ItemValue(String item_code, Object item_value) { |
|||
this.item_code = item_code; |
|||
this.item_value = item_value; |
|||
} |
|||
|
|||
public String getItem_code() { |
|||
return this.item_code; |
|||
} |
|||
|
|||
public void setItem_code(String item_code) { |
|||
this.item_code = item_code; |
|||
} |
|||
|
|||
public Object getItem_value() { |
|||
return this.item_value; |
|||
} |
|||
|
|||
public void setItem_value(Object item_value) { |
|||
this.item_value = item_value; |
|||
} |
|||
} |
|||
@ -0,0 +1,78 @@ |
|||
package org.nl.acs.opc; |
|||
|
|||
|
|||
import org.nl.modules.wql.exception.WDKException; |
|||
|
|||
import java.io.ByteArrayInputStream; |
|||
import java.io.ByteArrayOutputStream; |
|||
import java.io.ObjectInputStream; |
|||
import java.io.ObjectOutputStream; |
|||
import java.lang.reflect.InvocationTargetException; |
|||
import java.lang.reflect.Method; |
|||
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); |
|||
} |
|||
} |
|||
} |
|||
@ -0,0 +1,177 @@ |
|||
package org.nl.acs.opc; |
|||
|
|||
import org.jinterop.dcom.common.JIException; |
|||
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 org.springframework.beans.factory.annotation.Autowired; |
|||
|
|||
import java.util.*; |
|||
import java.util.concurrent.CompletableFuture; |
|||
import java.util.concurrent.ConcurrentHashMap; |
|||
import java.util.stream.Collectors; |
|||
public class OpcEntity { |
|||
@Autowired |
|||
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; |
|||
}))); |
|||
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; |
|||
} |
|||
} |
|||
@ -0,0 +1,22 @@ |
|||
package org.nl.acs.opc; |
|||
|
|||
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); |
|||
} |
|||
@ -0,0 +1,191 @@ |
|||
package org.nl.acs.opc; |
|||
|
|||
import cn.hutool.core.util.StrUtil; |
|||
import org.jinterop.dcom.common.JIException; |
|||
import org.nl.acs.auto.initial.ApplicationAutoInitial; |
|||
import org.nl.modules.common.exception.BadRequestException; |
|||
import org.openscada.opc.lib.common.NotConnectedException; |
|||
import org.openscada.opc.lib.da.Group; |
|||
import org.openscada.opc.lib.da.Server; |
|||
import org.openscada.opc.lib.da.UnknownGroupException; |
|||
import org.slf4j.Logger; |
|||
import org.slf4j.LoggerFactory; |
|||
import org.springframework.beans.factory.annotation.Autowired; |
|||
import org.springframework.stereotype.Service; |
|||
|
|||
import java.net.UnknownHostException; |
|||
import java.util.Collections; |
|||
import java.util.HashMap; |
|||
import java.util.Iterator; |
|||
import java.util.Map; |
|||
|
|||
/** |
|||
* @author ldjun |
|||
* @version 1.0 |
|||
* @date 2023年02月01日 11:27 |
|||
* @desc desc |
|||
*/ |
|||
@Service |
|||
public class OpcServerServiceImpl implements OpcServerService, ApplicationAutoInitial { |
|||
|
|||
private static final Logger log = LoggerFactory.getLogger(OpcServerServiceImpl.class); |
|||
@Autowired OpcServerManageService opcServerManageService; |
|||
Map<String, OpcServerManageDto> opcServerManageDtos = new HashMap(); |
|||
Map<String, Server> servers = Collections.synchronizedMap(new HashMap()); |
|||
Map<String, Group> groups = Collections.synchronizedMap(new HashMap()); |
|||
|
|||
public OpcServerServiceImpl() {} |
|||
|
|||
public void autoInitial() throws Exception { |
|||
this.reload(); |
|||
if (OpcConfig.auto_start_opc) { |
|||
Thread t = |
|||
new Thread() { |
|||
public void run() { |
|||
Iterator var1 = OpcServerServiceImpl.this.opcServerManageDtos.values().iterator(); |
|||
|
|||
while (var1.hasNext()) { |
|||
OpcServerManageDto dto = (OpcServerManageDto) var1.next(); |
|||
|
|||
try { |
|||
OpcServerServiceImpl.this.getServer(dto.getOpc_code()); |
|||
OpcServerServiceImpl.log.info("加载opc server {}", dto.getOpc_code()); |
|||
} catch (Exception var4) { |
|||
OpcServerServiceImpl.log.warn("启动无法载入servers", var4); |
|||
} |
|||
} |
|||
} |
|||
}; |
|||
t.start(); |
|||
} |
|||
} |
|||
|
|||
public synchronized void reload() { |
|||
this.opcServerManageDtos = this.opcServerManageService.queryAllServerMap(); |
|||
this.opcServerManageDtos = Collections.synchronizedMap(this.opcServerManageDtos); |
|||
} |
|||
|
|||
public void cleanGroups(String opcCode) { |
|||
Group group = (Group)this.groups.get(opcCode); |
|||
if (group != null) { |
|||
Server server = group.getServer(); |
|||
|
|||
try { |
|||
group.remove(); |
|||
} catch (JIException var5) { |
|||
var5.printStackTrace(); |
|||
} |
|||
|
|||
this.groups.remove(opcCode); |
|||
server.disconnect(); |
|||
this.servers.remove(opcCode); |
|||
} |
|||
|
|||
} |
|||
|
|||
public Group getServer(String code) throws Exception { |
|||
synchronized(this.buildLock(code)) { |
|||
Group group = null; |
|||
group = (Group)this.groups.get(code); |
|||
if (group != null) { |
|||
label68: { |
|||
Group var10000; |
|||
try { |
|||
if (!group.isActive()) { |
|||
break label68; |
|||
} |
|||
|
|||
var10000 = group; |
|||
} catch (JIException var14) { |
|||
log.error(code, var14); |
|||
break label68; |
|||
} |
|||
|
|||
return var10000; |
|||
} |
|||
} |
|||
|
|||
Server server = (Server)this.servers.get(code); |
|||
boolean needcreate = false; |
|||
String groupName = code; |
|||
if (server == null) { |
|||
needcreate = true; |
|||
} else { |
|||
try { |
|||
group = server.findGroup(groupName); |
|||
} catch (UnknownHostException | JIException | UnknownGroupException | NotConnectedException | IllegalArgumentException var13) { |
|||
log.error(code, var13); |
|||
needcreate = true; |
|||
} |
|||
} |
|||
|
|||
if (needcreate) { |
|||
OpcServerManageDto dto = (OpcServerManageDto)this.opcServerManageDtos.get(code); |
|||
if (dto == null) { |
|||
throw new BadRequestException(code + "{} 不存在"); |
|||
} |
|||
|
|||
if (server == null) { |
|||
server = OpcServerUtl.getServerWithOutException(dto.getOpc_host(), StrUtil.trim(dto.getCls_id()), dto.getUser(), dto.getPassword(), StrUtil.trim(dto.getDomain())); |
|||
} |
|||
|
|||
try { |
|||
group = server.addGroup(groupName); |
|||
} catch (Exception var12) { |
|||
this.clearServer(code); |
|||
ThreadUtl.sleep(5000L); |
|||
log.warn("获取opc出错重新获取", code, var12); |
|||
server = OpcServerUtl.getServerWithOutException(dto.getOpc_host(), StrUtil.trim(dto.getCls_id()), dto.getUser(), dto.getPassword(), StrUtil.trim(dto.getDomain())); |
|||
|
|||
try { |
|||
group = server.addGroup(groupName); |
|||
} catch (Exception var11) { |
|||
var11.printStackTrace(); |
|||
} |
|||
} |
|||
|
|||
this.servers.put(code, server); |
|||
this.groups.put(code, group); |
|||
} |
|||
|
|||
return group; |
|||
} |
|||
} |
|||
|
|||
public void clearServer(String code) { |
|||
synchronized(this.buildLock(code)) { |
|||
try { |
|||
Server server = (Server)this.servers.get(code); |
|||
server.disconnect(); |
|||
} catch (Exception var5) { |
|||
} |
|||
|
|||
this.servers.remove(code); |
|||
this.groups.remove(code); |
|||
} |
|||
} |
|||
|
|||
|
|||
public void writeInteger(String code, ItemValue... values) { |
|||
try { |
|||
Group group = this.getServer(code); |
|||
OpcUtl.writeValue(group, values); |
|||
} catch (Exception var4) { |
|||
this.clearServer(code); |
|||
log.warn("写入出错opc server {} 重新加载", code, var4); |
|||
ThreadUtl.sleep(5000L); |
|||
// throw var4;
|
|||
} |
|||
} |
|||
|
|||
private String buildLock(String key) { |
|||
if (StrUtil.isEmpty(key)) { |
|||
key = ""; |
|||
} |
|||
StringBuilder builder = new StringBuilder(); |
|||
builder.append("OpcServerService."); |
|||
builder.append(key); |
|||
String lock = builder.toString().intern(); |
|||
return lock; |
|||
} |
|||
} |
|||
@ -1,5 +1,5 @@ |
|||
package org.nl.acs.opc; |
|||
|
|||
public class OpcStartTag { |
|||
public static boolean is_run = true; |
|||
public static boolean is_run = false; |
|||
} |
|||
|
|||
@ -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); |
|||
} |
|||
} |
|||
} |
|||
|
|||
@ -0,0 +1,23 @@ |
|||
package org.nl.modules.logging; |
|||
|
|||
import ch.qos.logback.core.PropertyDefinerBase; |
|||
import org.springframework.stereotype.Component; |
|||
|
|||
// 通过实现logback的PropertyDefinerBase方法,动态定义logback配置中的变量
|
|||
@Component |
|||
public class DeviceCodeDir extends PropertyDefinerBase { |
|||
|
|||
String deviceCodeDir = ""; |
|||
public void setPropertyValue(String deviceCode) { |
|||
deviceCodeDir = deviceCode ; |
|||
} |
|||
|
|||
@Override |
|||
public String getPropertyValue() { |
|||
if (deviceCodeDir == ""){ |
|||
deviceCodeDir = "默认"; |
|||
} |
|||
return "默认"; |
|||
} |
|||
|
|||
} |
|||
Loading…
Reference in new issue