HikariThings

外观

Java 实现 OPC UA 通信（Eclipse Milo） ​

原稿给出学习资料与实现方向（Milo）；下文按官方示例与常用实践补全具体实现步骤（连接 / 浏览 / 读 / 写 / 订阅 / 简易服务端）。
版本说明：示例面向 Milo 1.1.x（JDK 17+）。1.2 起部分订阅 API 有迁移，升级请对照 Milo Client Wiki。

1、资料（原稿） ​

资源链接
博客标签（OPC 系列）https://www.cnblogs.com/ioufev/tag/OPC/
视频讲解https://www.bilibili.com/video/BV1TR4y1Q71g/
Eclipse Milo 源码https://github.com/eclipse/milo（现主仓多为 eclipse-milo/milo）

另可配合 UaExpert 等客户端先连上目标服务器，确认 namespaceIndex、节点 identifier，再写 Java。

2、如何实现（原稿 + 展开） ​

可以使用开源库 Milo 实现 OPC UA 的客户端和服务端。

工业现场常见形态：

text
PLC / SCADA / 专有 OPC UA Server
        ↑ opc.tcp://...
   Java（Milo Client）  ← 你自研网关、采集服务、或二开模块
        ↓ MQTT / HTTP / Kafka …
   ThingsBoard（设备遥测 / 属性）

若不想自研协议栈，也可用 ThingsBoard IoT Gateway 的 OPC-UA Connector 直接采数上报平台；本文聚焦 Java + Milo 的可编程实现。

2.1、Maven 依赖（推荐 Milo 1.1.6） ​

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

<!-- 若要自己起 OPC UA Server -->
<dependency>
  <groupId>org.eclipse.milo</groupId>
  <artifactId>milo-sdk-server</artifactId>
  <version>1.1.6</version>
</dependency>

旧教程常见 0.6.x（artifact 多为 sdk-client）。包名与订阅 API 与 1.x 有差异；跟博客代码时注意版本对齐。官方示例见仓库 milo-examples。

2.2、联调服务器 ​

本地没有设备时，可先用公开演示或 Docker：

  • Milo 相关演示端点（以当时是否在线为准）：opc.tcp://milo.digitalpetri.com:62541/milo
  • Gateway 文档常用演示镜像：docker run -it -p 4840:4840 thingsboard/tb-gw-opcua-server:latest

用 UaExpert 匿名连接，记下要读的节点，例如 namespace=2，identifier=Dynamic/RandomDouble。

2.3、创建并连接客户端 ​

核心思路：OpcUaClient.create(endpointUrl, …) 拉取 Endpoint → 选策略/安全 → connect() 建 SecureChannel 与 Session。

java
import org.eclipse.milo.opcua.sdk.client.OpcUaClient;
import org.eclipse.milo.opcua.sdk.client.identity.AnonymousProvider;
import org.eclipse.milo.opcua.stack.core.security.SecurityPolicy;
import org.eclipse.milo.opcua.stack.core.types.builtin.LocalizedText;

public class OpcUaClientFactory {

    public static OpcUaClient createAndConnect(String endpointUrl)
            throws Exception {

        // 演示：选 SecurityPolicy.None + 匿名。
        // 也可 OpcUaClient.create(endpointUrl) 自动挑「无安全+匿名」端点。
        OpcUaClient client = OpcUaClient.create(
                endpointUrl,
                endpoints -> endpoints.stream()
                        .filter(e -> e.getSecurityPolicyUri()
                                .equals(SecurityPolicy.None.getUri()))
                        .findFirst(),
                configBuilder -> configBuilder
                        .setApplicationName(LocalizedText.english("hikari-opcua-client"))
                        .setApplicationUri("urn:hikari:opcua:client")
                        .setIdentityProvider(new AnonymousProvider()) // 默认即匿名，可省略
                        .build()
        );

        client.connect().get();
        return client;
    }

    public static void main(String[] args) throws Exception {
        String url = "opc.tcp://127.0.0.1:4840";
        OpcUaClient client = createAndConnect(url);
        try {
            // 读 / 写 / 订阅 …
        } finally {
            client.disconnect().get();
        }
    }
}

要点：

  • 一个应用进程对一台 OPC UA Server 通常共用一个 OpcUaClient，断线由 Milo 尝试恢复 Session / 订阅；业务侧在停机时 disconnect()。
  • Endpoint 过滤要与服务器实际安全策略一致，否则连不上。

2.4、浏览地址空间（找 NodeId） ​

java
import org.eclipse.milo.opcua.sdk.client.nodes.UaNode;
import org.eclipse.milo.opcua.stack.core.Identifiers;
import org.eclipse.milo.opcua.stack.core.types.builtin.NodeId;

import java.util.List;

public static void browseRoot(OpcUaClient client) throws Exception {
    UaNode root = client.getAddressSpace()
            .getNode(Identifiers.ObjectsFolder);

    List<? extends UaNode> children =
            client.getAddressSpace().browseNodes(root);

    for (UaNode child : children) {
        NodeId id = child.getNodeId();
        System.out.printf("browse: %s  %s%n",
                child.getBrowseName().getName(), id);
    }
}

现场节点命名因厂商而异（通道.设备.点位）。以 UaExpert 看到的 NodeId 为准，不要猜 namespace。

2.5、读节点 ​

java
import org.eclipse.milo.opcua.stack.core.types.builtin.DataValue;
import org.eclipse.milo.opcua.stack.core.types.builtin.NodeId;
import org.eclipse.milo.opcua.stack.core.types.enumerated.TimestampsToReturn;

public static Object readValue(OpcUaClient client,
                               int namespaceIndex,
                               String identifier) throws Exception {
    NodeId nodeId = new NodeId(namespaceIndex, identifier);
    DataValue dataValue = client.readValue(
            0.0, TimestampsToReturn.Both, nodeId).get();

    if (dataValue.getStatusCode() == null
            || !dataValue.getStatusCode().isGood()) {
        throw new IllegalStateException(
                "read failed: " + dataValue.getStatusCode());
    }
    Object raw = dataValue.getValue().getValue();
    System.out.println(identifier + " = " + raw
            + "  ts=" + dataValue.getSourceTime());
    return raw;
}

// 例：readValue(client, 2, "Dynamic/RandomDouble");

批量读可用 client.readValues(maxAge, timestamps, List<NodeId>)，减少往返。

2.6、写节点 ​

写入类型必须与服务器节点数据类型一致（Float / Double / Int32 等），否则 Status 非 Good。

java
import org.eclipse.milo.opcua.stack.core.types.builtin.DataValue;
import org.eclipse.milo.opcua.stack.core.types.builtin.StatusCode;
import org.eclipse.milo.opcua.stack.core.types.builtin.Variant;

public static void writeValue(OpcUaClient client,
                              int namespaceIndex,
                              String identifier,
                              Object javaValue) throws Exception {
    NodeId nodeId = new NodeId(namespaceIndex, identifier);
    DataValue dv = new DataValue(new Variant(javaValue), null, null);
    StatusCode status = client.writeValue(nodeId, dv).get();
    if (!status.isGood()) {
        throw new IllegalStateException("write failed: " + status);
    }
}

// 例：writeValue(client, 2, "Demo.Setpoint", 26.5f);

2.7、订阅（推荐采集方式） ​

轮询 read 简单但费带宽；产线更常用 Subscription + MonitoredItem：服务器在值变化（或按采样周期）推送。

java
import org.eclipse.milo.opcua.sdk.client.api.subscriptions.UaMonitoredItem;
import org.eclipse.milo.opcua.sdk.client.api.subscriptions.UaSubscription;
import org.eclipse.milo.opcua.stack.core.AttributeId;
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;

import java.util.Collections;
import java.util.List;
import java.util.concurrent.atomic.AtomicInteger;

public static void subscribe(OpcUaClient client,
                             int namespaceIndex,
                             String identifier) throws Exception {

    UaSubscription subscription =
            client.getSubscriptionManager()
                    .createSubscription(1000.0)  // 发布间隔 ms
                    .get();

    NodeId nodeId = new NodeId(namespaceIndex, identifier);
    ReadValueId readValueId = new ReadValueId(
            nodeId,
            AttributeId.Value.uid(),
            null,
            null
    );

    AtomicInteger handle = new AtomicInteger(1);
    MonitoringParameters parameters = new MonitoringParameters(
            UInteger.valueOf(handle.getAndIncrement()),
            1000.0,                 // 采样间隔 ms
            null,                   // filter
            UInteger.valueOf(10),   // 队列长度
            true                    // discard oldest
    );

    MonitoredItemCreateRequest request =
            new MonitoredItemCreateRequest(
                    readValueId, MonitoringMode.Reporting, parameters);

    List<UaMonitoredItem> items = subscription.createMonitoredItems(
            TimestampsToReturn.Both,
            Collections.singletonList(request),
            (item, id) -> item.setValueConsumer((monitoredItem, value) -> {
                System.out.println(
                        "notify " + monitoredItem.getReadValueId().getNodeId()
                                + " => " + value.getValue());
                // 这里可转成 MQTT 上报 ThingsBoard
            })
    ).get();

    for (UaMonitoredItem item : items) {
        System.out.println("monitored status=" + item.getStatusCode());
    }

    // 演示：阻塞保持订阅；正式环境用生命周期托管，勿 sleep 死循环
    Thread.sleep(60_000);
}

多点批量订阅：为每个 identifier 建一条 MonitoredItemCreateRequest，一次 createMonitoredItems。

2.8、安全与身份（生产必看） ​

方式说明
Anonymous仅适合演示 / 内网封闭环境
用户名密码UsernameProvider（具体类名随 Milo 版本）
证书应用证书 + 信任对方证书；策略如 Basic256Sha256

生产建议：独立应用证书、明确信任列表、最小权限账户；拒绝策略不匹配的 Endpoint。证书被拒时查服务器 rejected 目录并完成信任 reciprocal。

2.9、简易 OPC UA 服务端（Milo Server） ​

需要自己模拟点位或对外提供 UA 接口时，引入 milo-sdk-server，参考仓库 milo-examples/server-examples：

  1. 创建 OpcUaServer / 命名空间
  2. 在地址空间中挂 UaVariableNode（温度、设定值等）
  3. 绑定 tcp Endpoint（如 opc.tcp://0.0.0.0:4840）
  4. server.startup()

客户端连本机该端口即可联调读写订阅。完整样例较长，建议直接克隆 Milo 仓库跑官方 Example，再改成业务命名空间。

2.10、接到 ThingsBoard 的两种路径 ​

路径 A — 自研（本文）

text
Milo 订阅回调
  → 组装 JSON { "temperature": 26.5, ... }
  → MQTT 发布到 TB 设备遥测主题
     （或 HTTP Transport / 规则链）

适合：要深度定制映射、多协议汇聚、或嵌入现有 Java 服务。

路径 B — ThingsBoard IoT Gateway OPC-UA 连接器

在 Gateway 中启用 type: opcua，配置 server.url、安全策略、mapping（设备节点 / 属性 / 遥测路径）。官方说明：
https://thingsboard.io/docs/iot-gateway/config/opc-ua/

适合：快速把现有 OPC UA Server 采进平台，少写代码。

3、实现检查清单 ​

  1. 用 UaExpert 能连上，并抄对 NodeId
  2. Maven 坐标与 JDK 与所选 Milo 大版本一致
  3. Endpoint 安全策略过滤正确，connect() 成功
  4. 读：Status Good；写：类型匹配
  5. 订阅：createMonitoredItems 成功，回调有值
  6. 上报 TB：设备令牌 / 主题正确；大包注意 MQTT 解码上限

4、常见问题 ​

现象排查
连不上URL、端口、防火墙；安全策略/证书不匹配
读到 null / BadNodeId 错；无读权限；节点未激活
写失败Java 类型与 UA 数据类型不一致
订阅无回调采样/发布间隔；MonitoringMode；服务器不支持该项
依赖冲突Netty 等版本被 Spring Boot 覆盖，对齐 Milo 传递依赖

补充（辅） ​

  • 官方示例目录：milo-examples/client-examples（Read / Write / Subscription / Browse）。
  • 中文系列可对照原稿博客标签逐步跟做，再迁到你现场的 NodeId。
  • ThingsBoard 本体默认是 MQTT/HTTP 等接入；OPC UA 通常停在边缘（自研或 Gateway），再转成平台认识的遥测。

相关文章 ​

物联网平台 · 行业方案 · 硬件产品
让物联网交付更简单

产品与服务

学习与资讯

关于

技术咨询

平台选型、私有化交付与现场方案，欢迎通过社区或联系页沟通。

前往联系

© 2024–2026 HikariThings. All rights reserved.

HikariThings 相关商标归其权利人所有；本站内容用于工程实践与交流。