Java实现OPC UA断连重连和数据监听

一时失言乱红尘 2024-03-17 14:36 137阅读 0赞

## 简介 ##

本文将以Java语言为例，详细解读如何在OPC UA通信中实现断连重连和数据监听的技术。首先，分析了为何断连重连和数据监听在OPC UA应用中至关重要，以及传统方法的局限性。随后，引入了Java的开源框架和库，如Eclipse Milo和Apache Camel，以优雅且高效的方式处理连接管理和数据流。同时，结合实际案例，详细演示了如何利用这些技术代码实现OPC UA断连重连和数据监听，从而实现系统的稳定性和实时性。无论您是OPC UA初学者还是有一定经验的开发者，本文都将为您提供宝贵的技术指导，助力您实现高效稳定的OPC UA通信。

## 引入依赖 ##

首先在maven项目中，引入**org.eclipse.milo**jar包依赖

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

**Milo库**  
org.eclipse.milo是一个基于Java的开源OPC UA（开放型通讯联盟）实现。OPC UA是一种用于工业自动化领域的开放标准通信协议，它提供了可互操作的数据交换和设备管理能力。

org.eclipse.milo项目旨在提供一个完善的OPC UA实现，以便开发者可以轻松地创建和管理OPC UA服务器和客户端。它的设计目标是在性能和功能方面提供高度的可扩展性和灵活性。

该项目基于Eclipse IoT项目中的Eclipse Milo子项目进行开发和维护。Eclipse Milo提供一系列的OPC UA库和工具，可以帮助开发者在Java平台上构建OPC UA应用程序。org.eclipse.milo扩展了Eclipse Milo，提供了更多的功能和集成选项。

通过org.eclipse.milo，开发者可以轻松地创建基于OPC UA的应用程序，包括OPC UA服务器和客户端。它提供了一组API，可以处理与OPC UA通信相关的任务，如创建和管理节点，读写变量值，订阅和发布事件等。

org.eclipse.milo还提供了一些示例应用程序和工具，可以帮助开发者入门并快速开始开发OPC UA应用程序。

总结来说，org.eclipse.milo是一个功能强大的基于Java的OPC UA实现，为开发者提供了构建和管理OPC UA应用程序的工具和API。它是一个开源项目，可以根据需要进行定制和扩展。

## OPC UA断开重连 ##

创建订阅事件监听器SubscriptionListener实现implements UaSubscriptionManager.SubscriptionListener 接口的方法，代码如下：

import com.tarzan.opcua.util.OpcUaUtil;
    import lombok.extern.slf4j.Slf4j;
    import org.eclipse.milo.opcua.sdk.client.OpcUaClient;
    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.stack.core.UaException;
    import org.eclipse.milo.opcua.stack.core.types.builtin.DateTime;
    import org.eclipse.milo.opcua.stack.core.types.builtin.StatusCode;
    
    import java.util.Set;
    
    
    /**
     * @author tarzan
     */
    @Slf4j
    public class SubscriptionListener implements UaSubscriptionManager.SubscriptionListener {
        
    
    
        private Set<String> keys;
        private OpcUaClient client;
    
        public SubscriptionListener(Set<String> keys, OpcUaClient client) {
        
            this.keys = keys;
            this.client=client;
        }
    
        @Override
        public void onKeepAlive(UaSubscription subscription, DateTime publishTime) {
        
            log.info("onKeepAlive");
        }
    
        @Override
        public void onStatusChanged(UaSubscription subscription, StatusCode status) {
        
            log.info("onStatusChanged");
        }
    
        @Override
        public void onPublishFailure(UaException exception) {
        
            log.info("onPublishFailure");
        }
    
        @Override
        public void onNotificationDataLost(UaSubscription subscription) {
        
            log.info("onNotificationDataLost");
        }
    
        /**
         * 重连时 尝试恢复之前的订阅失败时 会调用此方法
         * @param uaSubscription 订阅
         * @param statusCode 状态
         */
        @Override
        public void onSubscriptionTransferFailed(UaSubscription uaSubscription, StatusCode statusCode) {
        
            log.info("恢复订阅失败 需要重新订阅");
            //在回调方法中重新订阅
            OpcUaUtil.handlerNode(keys,client);
        }
    }

其中 keys是需要订阅点位集合，client是OpcUa客户端。 onSubscriptionTransferFailed订阅转移失败时触发，通过调用OpcUaUtil.handlerNode方法，重新批量订阅点位。handlerNode方法实现代码如下：

/**
         * 处理订阅业务
         *
         * @param keys
         */
        public static void handlerNode(Set<String> keys, OpcUaClient client) {
        
            try {
        
                //创建订阅
                ManagedSubscription subscription = ManagedSubscription.create(client);
                List<NodeId> nodeIdList = new ArrayList<>();
                for (String key : keys) {
        
                    nodeIdList.add(new NodeId(2, key));
                }
                //监听
                List<ManagedDataItem> dataItemList = subscription.createDataItems(nodeIdList);
                for (ManagedDataItem managedDataItem : dataItemList) {
        
                    managedDataItem.addDataValueListener(new MyDataValueListener());
                }
            } catch (Exception e) {
        
                e.printStackTrace();
            }
        }

其中 MyDataValueListener 是自定义的opc ua 数据值变化监听器。

## 数据监听 ##

创建一个MyDataValueListener 类，实现ManagedDataItem.DataValueListener接口的onDataValueReceived的方法。

import lombok.AllArgsConstructor;
    import lombok.extern.slf4j.Slf4j;
    import org.eclipse.milo.opcua.sdk.client.subscriptions.ManagedDataItem;
    import org.eclipse.milo.opcua.stack.core.types.builtin.DataValue;
    
    /**
     * @author tarzan
     */
    @Slf4j
    @AllArgsConstructor
    public class MyDataValueListener implements ManagedDataItem.DataValueListener {
        
    
    
        @Override
        public void onDataValueReceived(ManagedDataItem managedDataItem, DataValue dataValue) {
        
            log.info(managedDataItem.getNodeId().getIdentifier().toString() + ":" + dataValue.getValue().getValue());
            //do some something
        }
    }

## 订阅使用 ##

创建一个批量订阅的方法，传入需要订阅的点位名集合和OpcUaClient客户端

/**
         * 批量订阅
         *
         * @param keys 订阅
         * @throws Exception
         */
        public  void subscribeEvent(Set<String> keys,OpcUaClient client) throws Exception {
        
            final CountDownLatch eventLatch = new CountDownLatch(1);
            //处理订阅业务
            handlerNode(keys,client);
            //添加订阅监听器，用于处理断线重连后的订阅问题
            client.getSubscriptionManager().addSubscriptionListener(new SubscriptionListener(keys,client));
            //持续监听
            eventLatch.await();
        }

进一步封装方法，在service层，创建订阅方法，代码如下：

public void subscribe(Set<String> keys) throws Exception {
        
            OpcUaClient client= OpcUaUtil.createClient(endPointUrl);
            opcUaService.subscribeEvent(keys,client);
      }

然后通过controller接口，调用service层subscribe订阅方法，或者在其他业务层里调用即可。这样就可以实现，订阅断开重连和订阅点位值变化监听的功能了，代码比较简单，需要加入自己的业务处理，请在原有的代码上加以改动！下面补充一下，OpcUaUtil.createClient创建客户端的方法，代码如下：

/**
         * 方法描述: 创建客户端
         *
         * @param endPointUrl
         * @return {@link OpcUaClient}
         * @throws
         */
        public static OpcUaClient createClient(String endPointUrl){
        
            return createClient(endPointUrl,null,null);
        }
    
        /**
         * 方法描述: 创建客户端
         *
         * @param endPointUrl
         * @param username
         * @param password
         * @return {@link OpcUaClient}
         * @throws
         */
        public static OpcUaClient createClient(String endPointUrl,String username,String password){
        
            log.info(endPointUrl);
            try {
        
                //获取安全策略
                List<EndpointDescription> endpointDescription = DiscoveryClient.getEndpoints(endPointUrl).get();
                //过滤出一个自己需要的安全策略
                EndpointDescription endpoint = endpointDescription.stream()
                        .filter(e -> e.getSecurityPolicyUri().equals(SecurityPolicy.None.getUri()))
                        .findFirst().orElse(null);
                IdentityProvider identityProvider=new AnonymousProvider();
                if(!StringUtils.isEmpty(username)||!StringUtils.isEmpty(password)){
        
                    identityProvider=new UsernameProvider(username,password);
                }
                // 设置配置信息
                OpcUaClientConfig config = OpcUaClientConfig.builder()
                        // opc ua 自定义的名称
                        .setApplicationName(LocalizedText.english("plc"))
                        // 地址
                        .setApplicationUri(endPointUrl)
                        // 安全策略等配置
                        .setEndpoint(endpoint)
                        .setIdentityProvider(identityProvider)
                        //等待时间
                        .setRequestTimeout(UInteger.valueOf(5000))
                        .build();
                // 准备连接
                OpcUaClient opcClient =OpcUaClient.create(config);
                //开启连接
                opcClient.connect().get();
                log.info("连接成功。。。success");
                return opcClient;
            } catch (Exception e) {
        
                e.printStackTrace();
                log.error("======== opc connection fail ========");
            }
            return null;
        }

## 结语 ##

在我做的项目中，一开始使用的这种监听点位值变化的代码实现，但是我这边的需求是要求实时输出点位值。如果用监听值的办法的话，可会好久不输出，因为点位值没有变化，就不会触发监听事件方法。我这改成了定时一次性读取多个点位值的代码实现。关于断开重连的解决方案，因为我设置的是定时任务，1秒执行一次。在OpcUaUtil.createClient的方法上又加一个获取Opc Ua客户端的方法，当createClient方法返回为null时，我就再次调用创建客户端的方法，代码如下：

public  OpcUaClient getClient(){
        
            if(client==null){
        
                client= OpcUaUtil.createClient(endPointUrl);
            }
            return client;
        }

在service层，创建订阅方法修改，代码如下：

public void subscribe(Set<String> keys) throws Exception {
        
            OpcUaClient client= OpcUaUtil.getClient(endPointUrl);
            opcUaService.subscribeEvent(keys,client);
      }