OPC UA客户端多订阅多NodeId异常：仅接收每个订阅最后一个NodeId值

问题描述 #

我编写了一个OPC UA客户端代码，创建10个订阅，每个订阅处理3个NodeId。尽管已为每个NodeId创建了监控项，但每个订阅仅能接收到该订阅中最后一个NodeId的值。

原代码 #

import org.eclipse.milo.opcua.sdk.client.OpcUaClient;
import org.eclipse.milo.opcua.sdk.client.subscriptions.UaMonitoredItem;
import org.eclipse.milo.opcua.sdk.client.subscriptions.UaSubscription;
import org.eclipse.milo.opcua.stack.core.AttributeId;
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.builtin.QualifiedName;
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 org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.List;
import static org.eclipse.milo.opcua.stack.core.types.builtin.unsigned.Unsigned.uint;
import static com.google.common.collect.Lists.newArrayList;

public class MultipleSubscriptionExample {
    private static final Logger logger = LoggerFactory.getLogger(MultipleSubscriptionExample.class);

    public static void main(String[] args) throws Exception {
        OpcUaClient client = OpcUaClient.create("opc.tcp://KCT-L-610:53530/OPCUA/SimulationServer");
        client.connect().get();

        // 创建10个订阅，每个订阅处理3个标签
        for (int i = 0; i < 10; i++) {
            createSubscription(client, i);
        }

        // 休眠以让订阅运行并收集数据
        Thread.sleep(10000);
        client.disconnect().get();
    }

    private static void createSubscription(OpcUaClient client, int subscriptionIndex) throws Exception {
        // 每次创建新订阅
        UaSubscription subscription = client.getSubscriptionManager().createSubscription(1000.0).get();

        // 定义当前订阅要监控的3个节点
        List<ReadValueId> readValueIds = newArrayList(
            new ReadValueId(new NodeId(3, 1007 + subscriptionIndex * 3), AttributeId.Value.uid(), null, QualifiedName.NULL_VALUE),
            new ReadValueId(new NodeId(3, 1008 + subscriptionIndex * 3), AttributeId.Value.uid(), null, QualifiedName.NULL_VALUE),
            new ReadValueId(new NodeId(3, 1009 + subscriptionIndex * 3), AttributeId.Value.uid(), null, QualifiedName.NULL_VALUE)
        );

        // 创建订阅的监控参数
        MonitoringParameters parameters = new MonitoringParameters(
            subscription.nextClientHandle(),
            1000.0, // 采样间隔：1秒
            null,   // 无过滤器，使用默认
            uint(10),  // 队列大小
            true    // 队列满时丢弃最旧值
        );
        
        // 创建监控项请求
        List<MonitoredItemCreateRequest> requests = newArrayList();
        for (ReadValueId readValueId : readValueIds) {
            requests.add(new MonitoredItemCreateRequest(readValueId, MonitoringMode.Reporting, parameters));
        }

        // 创建监控项并设置值更新回调
        UaSubscription.ItemCreationCallback onItemCreated = (item, id) -> {
            item.setValueConsumer((item1, value) -> onSubscriptionValue(item1, value, subscriptionIndex));
        };

        // 创建监控项并处理结果
        List<UaMonitoredItem> items = subscription.createMonitoredItems(
            TimestampsToReturn.Both,
            requests,
            onItemCreated
        ).get();

        for (UaMonitoredItem item : items) {
            if (item.getStatusCode().isGood()) {
                logger.info("Subscription {}: 为nodeId={}创建监控项成功", subscriptionIndex, item.getReadValueId().getNodeId());
            } else {
                logger.warn("Subscription {}: 为nodeId={}创建监控项失败 (状态码={})", subscriptionIndex, item.getReadValueId().getNodeId(), item.getStatusCode());
            }
        }
    }

    private static void onSubscriptionValue(UaMonitoredItem item, DataValue value, int subscriptionIndex) {
        // 监控项值更新时触发此方法
        logger.info("Subscription {}: 收到nodeId={}的值: {}", subscriptionIndex, item.getReadValueId().getNodeId(), value.getValue());
    }
}

运行输出 #

14:56:45.402 [milo-shared-thread-pool-3] INFO  c.m.o.MultipleSubscriptionExample - Subscription 0: Received value for nodeId=NodeId{ns=3, id=1009}: Variant{value=0.0}
14:56:45.403 [milo-shared-thread-pool-3] INFO  c.m.o.MultipleSubscriptionExample - Subscription 0: Received value for nodeId=NodeId{ns=3, id=1009}: Variant{value=-1.618126}
14:56:45.403 [milo-shared-thread-pool-3] INFO  c.m.o.MultipleSubscriptionExample - Subscription 0: Received value for nodeId=NodeId{ns=3, id=1009}: Variant{value=-2.0}
14:56:45.413 [milo-shared-thread-pool-3] INFO  c.m.o.MultipleSubscriptionExample - Subscription 1: Received value for nodeId=NodeId{ns=3, id=1012}: Variant{value=0.1333334}
14:56:45.417 [milo-shared-thread-pool-3] INFO  c.m.o.MultipleSubscriptionExample - Subscription 1: Received value for nodeId=NodeId{ns=3, id=1012}: Variant{value=27.0}
14:56:45.417 [milo-shared-thread-pool-3] INFO  c.m.o.MultipleSubscriptionExample - Subscription 1: Received value for nodeId=NodeId{ns=3, id=1012}: Variant{value=0.0}
14:56:45.417 [milo-shared-thread-pool-3] INFO  c.m.o.MultipleSubscriptionExample - Subscription 1: Received value for nodeId=NodeId{ns=3, id=1012}: Variant{value=28.0}
14:56:45.427 [milo-shared-thread-pool-3] INFO  c.m.o.MultipleSubscriptionExample - Subscription 2: Received value for nodeId=NodeId{ns=3, id=1015}: Variant{value=-5.235019}
14:56:45.427 [milo-shared-thread-pool-3] INFO  c.m.o.MultipleSubscriptionExample - Subscription 2: Received value for nodeId=NodeId{ns=3, id=1015}: Variant{value=0.209057}
14:56:45.427 [milo-shared-thread-pool-3] INFO  c.m.o.MultipleSubscriptionExample - Subscription 2: Received value for nodeId=NodeId{ns=3, id=1015}: Variant{value=-4.472503}
14:56:45.427 [milo-shared-thread-pool-3] INFO  c.m.o.MultipleSubscriptionExample - Subscription 2: Received value for nodeId=NodeId{ns=3, id=1015}: Variant{value=0.0}

问题原因及解决方案 #

问题根源 #

所有监控项共用了同一个MonitoringParameters实例，且仅调用了一次subscription.nextClientHandle()生成客户端句柄。OPC UA服务器通过唯一的客户端句柄区分不同监控项，重复句柄会导致服务器覆盖之前的监控项配置，最终只有最后一个监控项的更新能被正确推送。

修改方案 #

在创建MonitoredItemCreateRequest的循环中，为每个监控项生成独立的MonitoringParameters，确保每个监控项的客户端句柄唯一：

关键修改代码段

// 修改创建监控项请求的逻辑
List<MonitoredItemCreateRequest> requests = newArrayList();
for (ReadValueId readValueId : readValueIds) {
    // 每个监控项使用独立的MonitoringParameters，获取新的客户端句柄
    MonitoringParameters itemParams = new MonitoringParameters(
        subscription.nextClientHandle(),
        1000.0,
        null,
        uint(10),
        true
    );
    requests.add(new MonitoredItemCreateRequest(readValueId, MonitoringMode.Reporting, itemParams));
}

完整修改后的createSubscription方法

private static void createSubscription(OpcUaClient client, int subscriptionIndex) throws Exception {
    UaSubscription subscription = client.getSubscriptionManager().createSubscription(1000.0).get();

    List<ReadValueId> readValueIds = newArrayList(
        new ReadValueId(new NodeId(3, 1007 + subscriptionIndex * 3), AttributeId.Value.uid(), null, QualifiedName.NULL_VALUE),
        new ReadValueId(new NodeId(3, 1008 + subscriptionIndex * 3), AttributeId.Value.uid(), null, QualifiedName.NULL_VALUE),
        new ReadValueId(new NodeId(3, 1009 + subscriptionIndex * 3), AttributeId.Value.uid(), null, QualifiedName.NULL_VALUE)
    );
    
    List<MonitoredItemCreateRequest> requests = newArrayList();
    for (ReadValueId readValueId : readValueIds) {
        // 每个监控项分配独立的监控参数，保证句柄唯一
        MonitoringParameters itemParams = new MonitoringParameters(
            subscription.nextClientHandle(),
            1000.0,
            null,
            uint(10),
            true
        );
        requests.add(new MonitoredItemCreateRequest(readValueId, MonitoringMode.Reporting, itemParams));
    }

    UaSubscription.ItemCreationCallback onItemCreated = (item, id) -> {
        item.setValueConsumer((item1, value) -> onSubscriptionValue(item1, value, subscriptionIndex));
    };

    List<UaMonitoredItem> items = subscription.createMonitoredItems(
        TimestampsToReturn.Both,
        requests,
        onItemCreated
    ).get();

    for (UaMonitoredItem item : items) {
        if (item.getStatusCode().isGood()) {
            logger.info("Subscription {}: 为nodeId={}创建监控项成功", subscriptionIndex, item.getReadValueId().getNodeId());
        } else {
            logger.warn("Subscription {}: 为nodeId={}创建监控项失败 (状态码={})", subscriptionIndex, item.getReadValueId().getNodeId(), item.getStatusCode());
        }
    }
}

修改后，每个订阅的3个NodeId都能正常接收值更新。

内容的提问来源于stack exchange，提问作者rxhulcm