如何提升Python中OPC UA与PLC的通信效率？

问题原因 #

核心问题是单变量逐个写入的请求开销：

• 每个set_value调用都会发起一次独立的OPC UA请求，包含TCP握手、报文传输、服务器响应等环节，这些开销会随请求数量线性累加。5个请求耗时20ms，100个请求自然会逼近2秒。
• OPC UA本身支持高效的批量读写操作，单个请求即可处理数十甚至上百个变量，能大幅降低请求往返的总开销。

另外，数据类型转换存在小问题（不影响速度，但有潜在风险）：np.uint16被强制转换为Int16类型，可能导致超出范围的值出错，但这不是速度慢的原因。

优化方案 #

1. 实现批量写入 #

修改自定义OpcClient类，增加批量写入方法，利用OPC UA客户端的批量接口（比如write_values）一次性处理多个变量：

# 在OpcClient类中新增方法
def set_values(self, node_value_pairs: list[tuple[Node, any]]) -> None:
    nodes = []
    data_values = []
    for node, value in node_value_pairs:
        nodes.append(node)
        data_values.append(get_uadatatype(value))
    # 调用客户端批量写入接口
    self.client.write_values(nodes, data_values)

调用方式改为：

opc.set_values([
    (self.statusword_node, get_statusnumber(self.statusword)),
    (self.actualspeed_node, self.actualspeed),
    (self.errorcode_node, self.errorcode),
    (self.motorcurrent_node, self.motorcurrent),
    (self.motortorque_node, self.motortorque)
])

批量写入能将多个变量的请求合并为一次，总耗时会从20ms（5个变量）降到几毫秒级，100个变量的写入时间也会控制在几十毫秒内。

2. 优化客户端配置 #

• 检查是否启用异步操作：如果使用的OPC UA库支持异步批量请求，可以进一步提升效率（不过同步批量已经能解决大部分问题）。
• 确认服务器端的批量请求限制：部分PLC的OPC UA服务器对单次批量请求的变量数量有上限，可查阅PLC文档调整。
• 调整报文大小：适当增大客户端的发送/接收缓冲区，减少分片传输的开销。

3. 修正数据类型转换（可选） #

将np.uint16映射到正确的UInt16类型，避免类型不匹配问题：

def get_uadatatype(value) -> ua.DataValue:
    '''checks the type of value and converts it into a ua.DataValue type'''
    if isinstance(value, bool):
        return ua.DataValue(ua.Variant(value, ua.VariantType.Boolean))

    elif isinstance(value, int):
        return ua.DataValue(ua.Variant(value, ua.VariantType.Int16))
    
    elif isinstance(value, np.uint16):
        return ua.DataValue(ua.Variant(value, ua.VariantType.UInt16))

    elif isinstance(value, float):
        return ua.DataValue(ua.Variant(value, ua.VariantType.Float))

    else:
        loginfo(f"Please import convert function for {type(value)}")
        raise ValueError

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