0
0

Java 001：通过OPC UA协议连接KepServerEx进行读、写、订阅操作

参考前辈的踩坑记录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

<localRepository>${MAVEN_HOME}/repoHyb</localRepository>

5.3 配置Maven镜像：从阿里云Maven镜像同步和下载jar

<mirror><!--配置具体的仓库的下载镜像 从阿里云Maven仓库下载-->
    <id>nexus-aliyun</id><!-- 此镜像的唯一标识符，用来区分不同的mirror元素 -->
    <mirrorOf>central</mirrorOf><!-- 对哪种仓库进行镜像，简单说就是替代哪个仓库-->
    <name>Nexus aliyun</name><!-- 镜像名称 -->
    <url>http://maven.aliyun.com/nexus/content/groups/public</url>
</mirror>

5.4 在Eclipse中设置Maven安装路径

5.5 在Eclipse中设置Maven的配置文件

第6步：以上环境搭建好后，就可以正式coding了。

6.1 在eclipse中创建maven项目，并在pom.xml中配置Milo客户端依赖：

<dependency>
    <groupId>org.eclipse.milo</groupId>
    <artifactId>sdk-client</artifactId>
    <version>0.6.3</version>
</dependency>
<dependency>
    <groupId>org.eclipse.milo</groupId>
    <artifactId>sdk-server</artifactId>
    <version>0.6.3</version>
</dependency>

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<? extends UaNode> 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<MonitoredItemCreateRequest> 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<String> key = new ArrayList<>();
            key.add("通道 1.设备 1.标记 4");
            key.add("通道 1.设备 1.标记 1");

            List<NodeId> nodeIdList = new ArrayList<>();
            for (String s : key) {
                nodeIdList.add(new NodeId(2, s));
            }

            //监听
            List<ManagedDataItem> 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

标签： java 物联网 rabbitmq
本文转载自: https://blog.csdn.net/qq_28530139/article/details/129545400
版权归原作者 qq_28530139 所有， 如有侵权，请联系我们删除。
登录后发布评论

“Java 001：通过OPC UA协议连接KepServerEx进行读、写、订阅操作”的评论:

还没有评论
关于作者
...
overfit同步小助手
文章同步
相关阅读

C#与C++交互开发系列（二十）：跨进程通信之共享内存（Shared Memory）

hadoop技术实验- 第4章 HBase 开发：使用Java操作HBase

智能对话微信前端小程序设计与开发

前端配置开发环境：新电脑配置前端开发环境，Vue开发环境配置的详细过程

SpringBoot集成MQTT服务器和客户端并通过WebSocket将消息实时发送到前端（本地搭建MQTT服务器）

idea | 搭建 SpringBoot 项目之配置 Maven

Python毕业设计选题：基于django+vue的全国民宿信息管理系统

文章导航