Node.js OPCUA如何在单个Subscription中批量添加多个Monitored Item

现有可运行代码 #

目前我已实现单个监控项的订阅逻辑，代码如下：

const itemToMonitor = {
    nodeId: resolveNodeId("ns=2;s=Channel1.Device1.temprature"),
    attributeId: AttributeIds.Value
};
const itemToMonitor2 = {
        nodeId: resolveNodeId("ns=2;s=Channel1.Device1.altitude"),
        attributeId: AttributeIds.Value
};
const monitoringParamaters = {
    samplingInterval: 1000,
    discardOldest: true
};
the_subscription.monitor(itemToMonitor, monitoringParamaters, TimestampsToReturn.Both, (err, monitoredItem) => {
    monitoredItem.on("changed", function (dataValue) {
        console.log("monitored item changed:  temprature = ", dataValue.value.value);
    });
});
the_subscription.monitor(itemToMonitor2, monitoringParamaters, TimestampsToReturn.Both, (err, monitoredItem) => {
    monitoredItem.on("changed", function (dataValue) {
        console.log("monitored item changed:  altitude = ", dataValue.value.value);
    });
});

期望实现的批量订阅方式 #

我希望能像以下伪代码这样，一次性批量订阅多个监控项，并在统一的回调中处理所有变更数据：

the_subscription.monitor({itemToMonitor,itemToMonitor2}, monitoringParamaters, TimestampsToReturn.Both, (err, monitoredItem) => {
            monitoredItem.on("changed", function (datavalues) {
                datavalues.each(){
                    // 批量处理所有变更数据
                }
            });
        });

.NET Core参考实现 #

我曾在.NET Core中实现过类似的批量订阅功能，参考代码如下：

_subscription.AddItems(_nodes);
_subscription.FastDataChangeCallback = new FastDataChangeNotificationEventHandler(DataChanged);
_session.AddSubscription(_subscription);
_subscription.Create();

private void DataChanged(Subscription subscription, DataChangeNotification notification, IList<string> stringTable)
{
  // 统一处理所有监控项的变更
}

可行性说明及实现方案 #

直接使用你伪代码中的monitor({itemToMonitor,itemToMonitor2})方式不可行，因为node-opcua的subscription.monitor()方法仅支持单个监控项配置作为第一个参数。但可以通过以下几种方式实现类似.NET Core的批量订阅效果：

方式1：循环创建+统一回调处理 #

将所有监控项整理为数组，循环调用monitor()，并绑定统一的变更处理逻辑：

// 整理监控项数组，附带节点名称用于识别
const monitoredItemConfigs = [
    { nodeId: resolveNodeId("ns=2;s=Channel1.Device1.temprature"), name: "temprature" },
    { nodeId: resolveNodeId("ns=2;s=Channel1.Device1.altitude"), name: "altitude" }
];

const monitoringParameters = {
    samplingInterval: 1000,
    discardOldest: true
};

// 统一的变更处理函数
const handleDataChange = (itemName, dataValue) => {
    console.log(`monitored item changed: ${itemName} = `, dataValue.value.value);
    // 此处可添加统一的数据处理逻辑
};

// 循环创建监控项
monitoredItemConfigs.forEach(config => {
    const itemToMonitor = {
        nodeId: config.nodeId,
        attributeId: AttributeIds.Value
    };
    the_subscription.monitor(itemToMonitor, monitoringParameters, TimestampsToReturn.Both, (err, monitoredItem) => {
        if (err) {
            console.error(`监控${config.name}失败:`, err);
            return;
        }
        monitoredItem.on("changed", (dataValue) => handleDataChange(config.name, dataValue));
    });
});

方式2：监听Subscription的dataChange事件 #

node-opcua的Subscription对象自带dataChange事件，订阅内任何监控项变更时都会触发该事件，回调会返回所有变更的监控项数据，与.NET Core的FastDataChangeCallback逻辑最接近：

// 先创建所有监控项
const itemToMonitor = { nodeId: resolveNodeId("ns=2;s=Channel1.Device1.temprature"), attributeId: AttributeIds.Value };
const itemToMonitor2 = { nodeId: resolveNodeId("ns=2;s=Channel1.Device1.altitude"), attributeId: AttributeIds.Value };

const monitoringParameters = { samplingInterval: 1000, discardOldest: true };

the_subscription.monitor(itemToMonitor, monitoringParameters, TimestampsToReturn.Both);
the_subscription.monitor(itemToMonitor2, monitoringParameters, TimestampsToReturn.Both);

// 批量处理所有变更
the_subscription.on("dataChange", (changedItems) => {
    changedItems.forEach(item => {
        const nodeIdStr = item.nodeId.toString();
        // 通过nodeId识别具体节点
        if (nodeIdStr === "ns=2;s=Channel1.Device1.temprature") {
            console.log("temprature changed: ", item.dataValue.value.value);
        } else if (nodeIdStr === "ns=2;s=Channel1.Device1.altitude") {
            console.log("altitude changed: ", item.dataValue.value.value);
        }
    });
});

方式3：高效批量创建（适合大量节点） #

如果需要监控大量节点，推荐使用subscription.createMonitoredItems()方法，这是底层OPC UA协议的批量创建操作，比循环调用monitor()更高效：

const batchItems = [
    {
        itemToMonitor: { nodeId: resolveNodeId("ns=2;s=Channel1.Device1.temprature"), attributeId: AttributeIds.Value },
        monitoringMode: MonitoringMode.Reporting,
        requestedParameters: { samplingInterval: 1000, discardOldest: true }
    },
    {
        itemToMonitor: { nodeId: resolveNodeId("ns=2;s=Channel1.Device1.altitude"), attributeId: AttributeIds.Value },
        monitoringMode: MonitoringMode.Reporting,
        requestedParameters: { samplingInterval: 1000, discardOldest: true }
    }
];

the_subscription.createMonitoredItems(
    TimestampsToReturn.Both,
    batchItems,
    (err, results) => {
        if (err) {
            console.error("批量创建监控项失败:", err);
            return;
        }
        results.forEach((result, index) => {
            if (result.statusCode.isGood()) {
                result.monitoredItem.on("changed", (dataValue) => {
                    console.log(`第${index+1}个监控项变更:`, dataValue.value.value);
                });
            } else {
                console.error(`第${index+1}个监控项创建失败:`, result.statusCode.toString());
            }
        });
    }
);

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