Java 001：通过OPC UA协议连接KepServerEx进行读、写、订阅操作
作者：mmseoamin日期：2023-12-25
参考前辈的踩坑记录https://blog.csdn.net/weixin_45411740/article/details/124275985?spm=1001.2014.3001.5502，我Hyb在2023-3-15调通了自己的JavaOpcUaDemo。具体Java代码和KepServerEX延时补丁都在资源中。
第1步：安装激活KepServer，补丁在资源中，不详述。
第2步：在KepServer中做OpcUa配置。
2.1 先看桌面右下角有没EX图标
2.2 如果没有EX图标，要去开始菜单中找到KEPServerEX 6 Administration，点击打开它后，桌面右下角就会出现EX图标：
2.3 右键点击EX图标，选择“OPC UA 配置”，设置过程看这个动图：（一定要关注动图中192.168.10.60这个ip及端口的设置，它要与你代码中一致！）
第3步：在KepServer中设置用户名密码。
3.1 再一次点桌面右下角的EX图标，选择“设置”，点“用户管理器”，在Administrators右键选择“添加用户”。见图操作：
第4步：在KepServer中设置opcua匿名账户登录。
在KepServerEX界面，在“项目”右键，选择“属性”，选择“OPC UA”。在“客户端会话”、“允许匿名登录”中，选择“是”。如图：
以上KepServerEX设置完成后，就可以构建Java项目了。
第5步：首先搭建环境。下载Eclipse、配置Maven
上图分别对应：
5.1 从Maven下载安装包：apache-maven-3.9.0-bin.zip
5.2 配置Maven本地仓库在：conf/settings.xml
${MAVEN_HOME}/repoHyb
5.3 配置Maven镜像：从阿里云Maven镜像同步和下载jar
nexus-aliyun central Nexus aliyun http://maven.aliyun.com/nexus/content/groups/public
5.4 在Eclipse中设置Maven安装路径
5.5 在Eclipse中设置Maven的配置文件
第6步：以上环境搭建好后，就可以正式coding了。
6.1 在eclipse中创建maven项目，并在pom.xml中配置Milo客户端依赖：
org.eclipse.milo sdk-client0.6.3 org.eclipse.milo sdk-server0.6.3
Milo是Eclipse旗下的一个物联网的项目，是一个高性能的OPC UA栈，提供了一组客户端和服务端的API，支持对实时数据的访问，监控，报警，订阅数据，支持事件，历史数据访问，和数据建模。官网https://projects.eclipse.org/projects/iot.milo
在Milo中大量的采用了java 8的新特性CompletableFuture来进行异步操作，Milo中有大量的操作都是直接返回CompletableFuture对象，所以JDK的版本必须要8.0，对CompletableFuture不太熟悉的可以先去了解CompletableFuture的相关概念在来看Milo的官方例子会轻松很多。
6.2 然后就是java代码，只有一个类TestOpcUA.java，代码全部在这一个java文件中。直接复制到你java文件中，不用再导入任何类，加个包名，直接可以运行。
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.util.ArrayList;
import java.util.List;
import java.util.Objects;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicInteger;
import org.eclipse.milo.opcua.sdk.client.OpcUaClient;
import org.eclipse.milo.opcua.sdk.client.api.identity.AnonymousProvider;
import org.eclipse.milo.opcua.sdk.client.api.subscriptions.UaSubscription;
import org.eclipse.milo.opcua.sdk.client.api.subscriptions.UaSubscriptionManager;
import org.eclipse.milo.opcua.sdk.client.nodes.UaNode;
import org.eclipse.milo.opcua.sdk.client.subscriptions.ManagedDataItem;
import org.eclipse.milo.opcua.sdk.client.subscriptions.ManagedSubscription;
import org.eclipse.milo.opcua.stack.core.AttributeId;
import org.eclipse.milo.opcua.stack.core.Identifiers;
import org.eclipse.milo.opcua.stack.core.UaException;
import org.eclipse.milo.opcua.stack.core.security.SecurityPolicy;
import org.eclipse.milo.opcua.stack.core.types.builtin.DataValue;
import org.eclipse.milo.opcua.stack.core.types.builtin.DateTime;
import org.eclipse.milo.opcua.stack.core.types.builtin.LocalizedText;
import org.eclipse.milo.opcua.stack.core.types.builtin.NodeId;
import org.eclipse.milo.opcua.stack.core.types.builtin.StatusCode;
import org.eclipse.milo.opcua.stack.core.types.builtin.Variant;
import org.eclipse.milo.opcua.stack.core.types.builtin.unsigned.UInteger;
import org.eclipse.milo.opcua.stack.core.types.enumerated.MonitoringMode;
import org.eclipse.milo.opcua.stack.core.types.enumerated.TimestampsToReturn;
import org.eclipse.milo.opcua.stack.core.types.structured.MonitoredItemCreateRequest;
import org.eclipse.milo.opcua.stack.core.types.structured.MonitoringParameters;
import org.eclipse.milo.opcua.stack.core.types.structured.ReadValueId;
public class TestOpcUA {
private final static String endPointUrl = "opc.tcp://192.168.10.60:49320";
/**
* 创建OPC UA客户端
*/
private static OpcUaClient createClient() throws Exception {
//opc ua服务端地址
Path securityTempDir = Paths.get(System.getProperty("java.io.tmpdir"), "security");
Files.createDirectories(securityTempDir);
if (!Files.exists(securityTempDir)) {
throw new Exception("unable to create security dir: " + securityTempDir);
}
return OpcUaClient.create(endPointUrl, endpoints -> endpoints.stream().filter(e -> e.getSecurityPolicyUri().equals(SecurityPolicy.None.getUri())).findFirst(), configBuilder -> configBuilder.setApplicationName(LocalizedText.english("eclipse milo opc-ua client")).setApplicationUri("urn:eclipse:milo:examples:client")
//访问方式
.setIdentityProvider(new AnonymousProvider()).setRequestTimeout(UInteger.valueOf(500)).build());
}
/**
* 遍历树形节点
*
* @param client OPC UA客户端
* @param uaNode 节点
* @throws Exception
*/
private static void browseNode(OpcUaClient client, UaNode uaNode) throws Exception {
List nodes;
if (uaNode == null) {
nodes = client.getAddressSpace().browseNodes(Identifiers.ObjectsFolder);
} else {
nodes = client.getAddressSpace().browseNodes(uaNode);
}
for (UaNode nd : nodes) {
//排除系统行性节点，这些系统性节点名称一般都是以"_"开头
if (Objects.requireNonNull(nd.getBrowseName().getName()).contains("_")) {
continue;
}
System.out.println("Node= " + nd.getBrowseName().getName());
browseNode(client, nd);
}
}
/**
* 读取节点数据
*
* @param client OPC UA客户端
* @throws Exception
*/
private static void readNode(OpcUaClient client) throws Exception {
int namespaceIndex = 2;
String identifier = "通道 1.设备 1.标记 1";
//节点
NodeId nodeId = new NodeId(namespaceIndex, identifier);
//读取节点数据
DataValue value = client.readValue(0.0, TimestampsToReturn.Neither, nodeId).get();
//标识符
identifier = String.valueOf(nodeId.getIdentifier());
System.out.println(identifier + ": " + String.valueOf(value.getValue().getValue()));
}
/**
* 写入节点数据
*
* @param client
* @throws Exception
*/
private static void writeNodeValue(OpcUaClient client) throws Exception {
//节点
NodeId nodeId = new NodeId(2, "通道 1.设备 1.标记 4");
Short i = 48;
//创建数据对象,此处的数据对象一定要定义类型，不然会出现类型错误，导致无法写入
DataValue nowValue = new DataValue(new Variant(i), null, null);
//写入节点数据
StatusCode statusCode = client.writeValue(nodeId, nowValue).join();
System.out.println("结果：" + statusCode.isGood());
}
/**
* 订阅(单个)
*
* @param client
* @throws Exception
*/
private static void subscribe(OpcUaClient client) throws Exception {
AtomicInteger a = new AtomicInteger();
//创建发布间隔1000ms的订阅对象
client.getSubscriptionManager().createSubscription(1000.0).thenAccept(t -> {
//节点
NodeId nodeId = new NodeId(2, "通道 1.设备 1.标记 1");
ReadValueId readValueId = new ReadValueId(nodeId, AttributeId.Value.uid(), null, null);
//创建监控的参数
MonitoringParameters parameters = new MonitoringParameters(UInteger.valueOf(a.getAndIncrement()), 1000.0, null, UInteger.valueOf(10), true);
//创建监控项请求
//该请求最后用于创建订阅。
MonitoredItemCreateRequest request = new MonitoredItemCreateRequest(readValueId, MonitoringMode.Reporting, parameters);
List requests = new ArrayList<>();
requests.add(request);
//创建监控项，并且注册变量值改变时候的回调函数。
t.createMonitoredItems(TimestampsToReturn.Both, requests, (item, id) -> item.setValueConsumer((it, val) -> {
System.out.println("nodeid :" + it.getReadValueId().getNodeId());
System.out.println("value :" + val.getValue().getValue());
}));
}).get();
//持续订阅
Thread.sleep(Long.MAX_VALUE);
}
/**
* 批量订阅
*
* @param client
* @throws Exception
*/
// private static void managedSubscriptionEvent(OpcUaClient client) throws Exception {
// final CountDownLatch eventLatch = new CountDownLatch(1);
//
// //处理订阅业务
// handlerNode(client);
//
// //持续监听
// eventLatch.await();
// }
/**
* 处理订阅业务
*
* @param client OPC UA客户端
*/
private static void handlerNode(OpcUaClient client) {
try {
//创建订阅
ManagedSubscription subscription = ManagedSubscription.create(client);
//你所需要订阅的key
List key = new ArrayList<>();
key.add("通道 1.设备 1.标记 4");
key.add("通道 1.设备 1.标记 1");
List nodeIdList = new ArrayList<>();
for (String s : key) {
nodeIdList.add(new NodeId(2, s));
}
//监听
List dataItemList = subscription.createDataItems(nodeIdList);
for (ManagedDataItem managedDataItem : dataItemList) {
managedDataItem.addDataValueListener((t) -> {
System.out.println(managedDataItem.getNodeId().getIdentifier().toString() + ":" + t.getValue().getValue().toString());
});
}
} catch (Exception e) {
e.printStackTrace();
}
}
/**
* 自定义订阅监听
*/
private static class CustomSubscriptionListener implements UaSubscriptionManager.SubscriptionListener {
private OpcUaClient client;
CustomSubscriptionListener(OpcUaClient client) {
this.client = client;
}
public void onKeepAlive(UaSubscription subscription, DateTime publishTime) {
System.out.println("onKeepAlive");
}
public void onStatusChanged(UaSubscription subscription, StatusCode status) {
System.out.println("onStatusChanged");
}
public void onPublishFailure(UaException exception) {
System.out.println("onPublishFailure");
}
public void onNotificationDataLost(UaSubscription subscription) {
System.out.println("onNotificationDataLost");
}
/**
* 重连时 尝试恢复之前的订阅失败时 会调用此方法
* @param uaSubscription 订阅
* @param statusCode 状态
*/
public void onSubscriptionTransferFailed(UaSubscription uaSubscription, StatusCode statusCode) {
System.out.println("恢复订阅失败 需要重新订阅");
//在回调方法中重新订阅
handlerNode(client);
}
}
/**
* 批量订阅
*
* @param client
* @throws Exception
*/
private static void managedSubscriptionEvent(OpcUaClient client) throws Exception {
final CountDownLatch eventLatch = new CountDownLatch(1);
//添加订阅监听器，用于处理断线重连后的订阅问题
client.getSubscriptionManager().addSubscriptionListener(new CustomSubscriptionListener(client));
//处理订阅业务
handlerNode(client);
//持续监听
eventLatch.await();
}
public static void main(String[] args) throws Exception {
OpcUaClient client = createClient();
client.connect().get();
// browseNode(client, null);
// readNode(client);
// writeNodeValue(client);
subscribe(client);
// managedSubscriptionEvent(client);
}
}
本文涉及的java项目代码和KepServerEX补丁下载地址：https://download.csdn.net/download/qq_28530139/87579334
Hyb✈✈✈本文结束，谢谢浏览【VX:m3920752】！个人资源的解压密码是: hyb
- 关注我们
- 扫描二维码
- 获取更多信息
- 精彩一手掌握
