多协议接入适配器

第 2 / 10 章

一、回顾与本章问题

上一章我把 scada-collect 的整体骨架立住了——5 张表搞定领域模型，TDengine 存时序，SSE 推实时，单实例部署。但骨架只是容器，真正把数据从 PLC 拉上来的，是这一章要讲的 4 个 Driver。

工厂现场的协议割裂有多严重？我上次去的那家注塑厂，4 台中央供料用 Modbus TCP，8 台注塑机用西门子 S7-1200，2 台冷干机走 OPC UA，新加的能耗网关推 MQTT Sparkplug B。四种协议、四套地址语法、四种连接参数。如果每接一种协议就从头写一遍"连接 → 读 → 写 → 重连"的代码，项目会变成一坨条件分支。所以第一步是抽象一个适配层，让上层只认一个接口。

二、行业对比：别人的适配层怎么设计

ThingsBoard 的做法是"传输层完全独立"：每种协议一个 TransportService，用 Netty 实现，通过 SPI（Java ServiceLoader）加载，协议之间完全隔离，加协议不改主仓代码。它的代价是抽象层级深——从 Channel 到 TransportService 到 DeviceProfile 到 RuleChain，新人上手要啃一周。而且 Netty Pipeline 处理的是字节流，对 OPC UA 这种基于 Session 的协议并不优雅，等于在 Netty 上又包了一层 Session 管理。

Ignition 走的是"Driver SDK 内聚"路线，每个 Driver 是一个独立模块，自带连接管理、地址解析、批量优化，对上层暴露统一的 readBlock/writeOnce 接口。它的 OPC UA Driver 实现了完整的订阅 + 死区 + 质量位处理，工业现场口碑很好。

我的取舍介于两者之间，更偏向 Ignition：用一个 Java 接口 + Spring @Component 注入，不做 SPI，不做 Netty。理由是项目规模摆在这——4 种协议，加一个新协议就是加一个实现类，没必要上 SPI 那套 META-INF/services 配置。IDE 点接口名能跳到所有实现，比 SPI 反射加载友好太多。

三、接口抽象：ScadaPlcDriver

接口本身极其克制，6 个方法撑死。代码在 yudao-module-scada/src/main/java/cn/iocoder/yudao/module/scada/framework/scada/driver/ScadaPlcDriver.java：

public interface ScadaPlcDriver {

    /** 该驱动支持的协议类型 */
    Integer getProtocolType();

    /** 建立到设备的连接，返回连接句柄；失败抛 RuntimeException */
    Object connect(ScadaDeviceDO device) throws Exception;

    /** 主动断开连接 */
    void disconnect(Object connHandle);

    /** 是否已连接，DriverManager 借出连接前自检 */
    boolean isConnected(Object connHandle);

    /**
     * 批量读取点位值。Driver 内部应按协议特性聚合读：
     * Modbus 按连续地址段聚合，OPC UA 用已建立的订阅缓存。
     */
    List<PointValue> readBatch(Object connHandle, List<ScadaPointDO> points) throws Exception;

    /** 单点写入，用于反向控制。只读协议（MQTT 订阅）无需实现 */
    default void writeOne(Object connHandle, ScadaPointDO point, Number value) throws Exception {
        throw new UnsupportedOperationException("Driver " + getProtocolType() + " 不支持写入");
    }
}

几个设计决策值得说道。第一，connect 返回 Object 而不是泛型 T。看起来不够优雅，但实际原因是连接句柄类型差异极大——Modbus 是 TCPMasterConnection、OPC UA 是 OpcUaClient、S7 是 PlcConnection、MQTT 是 MqttClient + 缓存 Map。用泛型会让 ScadaPlcDriverManager 的池化代码被迫变成 <T> 泛型类，池化逻辑和泛型耦合在一起很难看。返回 Object + Driver 内部强转，是务实的取舍。

第二，readBatch 而不是 readOne。这是工业场景的甜点——Modbus 一个报文能读 125 个寄存器，OPC UA 一个 Publish 能读一帧标签，按点位调度会浪费带宽。让 Driver 自己决定怎么聚合，上层只管"给一批点位，要一批值"。

第三，writeOne 用 default 方法抛 UnsupportedOperationException。MQTT Sparkplug 是订阅模式，没有反控语义，让它不用写一个永远抛异常的空实现。这个 default 是 Java 8 接口默认方法的正确用法——给"大多数协议都支持但少数不支持"的能力一个默认拒绝实现。

第四，不做"通用地址解析"。Modbus 是 HR40001、OPC UA 是 ns=2;s=Tag1、S7 是 DB1.DBX0.0，强行抽象成统一地址模型会变成字符串解析地狱。让每个 Driver 实现自己的 parseAddr，接口只规定"address 字段是字符串，怎么解析你看着办"。

四、池化管理：ScadaPlcDriverManager

有了接口，4 个 Driver 实现它，但谁来借连接、谁来归还、谁来重连？答案是 ScadaPlcDriverManager，它用 Commons-Pool2 做按设备 ID 池化。核心配置在 yudao-module-scada/src/main/java/cn/iocoder/yudao/module/scada/framework/scada/driver/ScadaPlcDriverManager.java：

GenericKeyedObjectPoolConfig<Object> cfg = new GenericKeyedObjectPoolConfig<>();
cfg.setMaxTotalPerKey(1);
cfg.setMaxIdlePerKey(1);
cfg.setMaxWait(Duration.ofMillis(2000));
cfg.setMinEvictableIdle(Duration.ofMinutes(5));
cfg.setBlockWhenExhausted(true);

maxTotalPerKey=1 是这一章最关键的一个数字。 一台设备一个连接，工业现场不允许并发报文。为什么？因为 PLC 的 Modbus 从站协议是"请求-响应"严格配对的，两个线程同时发 read 请求，PLC 返回的两帧响应可能交错，TCP 流里你分不清哪个响应对应哪个请求，解析直接乱套。OPC UA 虽然支持多 Session，但一个设备开多 Session 也浪费 PLC 侧连接配额。所以宁可让第二个任务排队等 2 秒（maxWait=2000），也不并发。

借用流程的代码也很简洁：

public List<PointValue> readBatch(ScadaDeviceDO device, List<ScadaPointDO> points) {
    ScadaPlcDriver driver = driverByProtocol.get(device.getProtocolType());
    Object conn = null;
    try {
        conn = connectionPool.borrowObject(device.getId());
        return driver.readBatch(conn, points);
    } catch (Exception e) {
        if (conn != null) {
            connectionPool.invalidateObject(device.getId(), conn);
            conn = null;
        }
        // 失败节流日志：1 秒内同一设备只打一次
        lastFailureLoggedAt.compute(device.getId(), (k, prev) -> {
            if (prev == null || now - prev > 1000) {
                log.warn("设备 {} 读取失败: {}", device.getCode(), e.getMessage());
                return now;
            }
            return prev;
        });
        throw new RuntimeException("设备 " + device.getCode() + " 读取失败: " + e.getMessage(), e);
    } finally {
        if (conn != null) connectionPool.returnObject(device.getId(), conn);
    }
}

这段代码的设计意图是：借连接失败时立即把连接标记为失效并从池中剔除（invalidateObject），下次借会触发 makeObject 重新 connect，自然形成重连。而失败日志用 ConcurrentHashMap.compute 做了 1 秒节流，避免断线期间日志刷屏——一个设备掉线后每秒会被扫一次，不打节流日志能堆出几万行。

@Autowired
public ScadaPlcDriverManager(List<ScadaPlcDriver> drivers) {
    for (ScadaPlcDriver d : drivers) {
        ScadaPlcDriver prev = driverByProtocol.put(d.getProtocolType(), d);
        if (prev != null) {
            log.warn("协议类型 {} 的 Driver 实现被覆盖：{} -> {}", ...);
        }
    }
}

Spring 会把所有 @Component 标注的 ScadaPlcDriver 实现收集成 List 注入，按 getProtocolType() 路由。如果同协议有两个实现，会 warn 提示覆盖——这就是为什么不用 SPI：SPI 重复实现是静默覆盖，排查起来要骂街。

五、四个 Driver 实现

Modbus TCP：j2mod + 批量合并

Modbus 是工业现场事实标准，几乎所有 PLC/变频器/数采网关都支持。地址模型分四类：HR（保持寄存器 40001+）、IR（输入寄存器 30001+）、coil（线圈 00001+）、DI（离散输入 10001+）。代码在 framework/scada/driver/ModbusTcpDriver.java，基于 j2mod（com.ghgande.j2mod）3.2.1。

批量合并是 Modbus 的甜点。一个 read 报文最多 125 个寄存器，如果点位列表里 100 个连续地址，一个报文就能搞定；如果一个个读，100 个报文，TCP 往返 100 次。我做的合并逻辑：

// 合并条件：相邻地址差 ≤ 4（贪心，避免过度合并导致中间寄存器被读浪费）
while (j + 1 < sorted.size()
        && parseAddr(sorted.get(j + 1).getAddress()) - parseAddr(sorted.get(j).getAddress()) <= 4) {
    j++;
}
int count = Math.min(end - start + 1, 125); // Modbus 单次最多 125 寄存器

为什么地址差阈值是 4？太大会把不相关的点位硬凑到一起，中间寄存器被白白读出来浪费带宽；太小则合并率低，退化成单点读。4 是经验值——典型 PLC 的温度、压力、流量参数往往按 4 字节（REAL32）一组排列，差 4 正好对齐一个 float。

地址解析把 HR40001 转成实际寄存器偏移 0：

private int parseAddr(String address) {
    String numPart = address.replaceAll("[^0-9]", "");
    int protocolAddr = Integer.parseInt(numPart);
    if (protocolAddr >= 40001) return protocolAddr - 40001;
    if (protocolAddr >= 30001) return protocolAddr - 30001;
    return protocolAddr;
}

Modbus 协议符号地址 40001 对应实际寄存器 0，这是协议规范，不是发明。

OPC UA：Eclipse Milo + 订阅兜底

OPC UA 是工业现场最高端的协议，支持订阅模式让 PLC 主动推值，不是 Driver 周期性 poll。基于 Eclipse Milo 0.6.13。代码在 framework/scada/driver/OpcUaDriver.java。

我用了"订阅缓存 + 兜底 Read"双模式：

// 先取订阅缓存：1 秒内的值直接用
long now = System.currentTimeMillis();
for (ScadaPointDO p : points) {
    CachedValue cached = oc.cache.get(p.getTag());
    if (cached != null && now - cached.ts < 1000) {
        result.add(PointValue.of(p.getId(), p.getTag(), cached.value));
    } else {
        result.add(null); // 占位，后面统一回退到 Read
    }
}
// 找出缓存未命中的，批量 Read
List<NodeId> nodeIds = new ArrayList<>(missing.size());
for (ScadaPointDO p : missing) nodeIds.add(parseNodeId(p.getAddress()));
List<DataValue> values = oc.client.readValues(0.0, null, nodeIds).get();

为什么这么设计？西门子 S7-1500 内置的 OPC UA Server 对订阅推送优化得很好，比 Modbus TCP 轮询节省 80% 带宽。但订阅可能因为网络抖动短暂失效，这时回退到 Read 服务一次性读，保证不漏。整个采集任务对同一设备的所有点位只需 1 个 Subscription，缓存是 ConcurrentHashMap<tag, CachedValue>，1 秒内有效。

S7：Apache PLC4X 统一 API

S7 是西门子 S7-200/300/400/1200/1500 系列专有协议。我选了 Apache PLC4X 0.13.0，没用 snap7-javase。代码在 framework/scada/driver/S7Driver.java。

为什么不用 snap7？snap7 是 LGPL 许可，商用闭源项目集成受限；PLC4X 是 Apache 顶级项目，许可干净，维护活跃。PLC4X 的另一个好处是统一 API——它把 S7、Modbus、OPC UA 都包成 PlcConnection，同样的 PlcReadRequest 接口。当然实际项目里我还是给每种协议写了独立 Driver，因为统一 API 抽象掉了协议特性（Modbus 的批量合并、OPC UA 的订阅），用 PLC4X 反而发挥不出协议优势。这里用 PLC4X 只是为了 S7 这一种协议的连接管理简洁。

地址语法 DB1.DBX0.0：DB1 是数据块号，DBX0.0 是第 0 个字节第 0 位（BOOL），DBW2 是第 2 个字（INT16），DBD4 是第 4 个双字（REAL32）。连接串是 s7://192.168.1.30?rack=0&slot=1，rack/slot 是 S7 协议的寻址参数，S7-1200 和 S7-1500 默认都是 rack=0 slot=1。

MQTT Sparkplug B：Paho + 订阅

MQTT Sparkplug B 是工业 MQTT 的标准化载荷格式，定义了 birth/death certificate、data message 等消息类型，让工业设备能"即插即用"接入 SCADA。代码在 framework/scada/driver/MqttSparkplugDriver.java，基于 Spring Integration MQTT + Eclipse Paho。

这个 Driver 是纯订阅模式——不主动读，订阅 topic 后等设备推值。订阅 topic 格式 spBv1.0/{group}/{edgeNode}/NDATA/+。重连后重新订阅在 connectComplete 回调里做：

client.setCallback(new MqttCallbackExtended() {
    @Override
    public void connectComplete(boolean reconnect, String serverURI) {
        String topic = String.format("spBv1.0/%s/%s/NDATA/+", group, edgeNode);
        client.subscribe(topic, 1);
    }
    @Override
    public void messageArrived(String topic, MqttMessage message) {
        Map<String, Object> data = gson.fromJson(payload, Map.class);
        for (Map.Entry<String, Object> e : data.entrySet()) {
            if (e.getValue() instanceof Number n) {
                cache.put(e.getKey(), BigDecimal.valueOf(n.doubleValue()));
            }
        }
    }
});

注意 connectComplete 的 reconnect 参数——Paho 配了 setAutomaticReconnect(true)，断线后会自动重连并触发这个回调，我们在这里重新订阅，保证 session 恢复后订阅不丢。cleanSession=false 配合，让 broker 保留离线消息。

读的时候从缓存取，10 秒内有效，超时算坏值——因为 MQTT 是事件驱动，如果一个点位 10 秒没新值，多半是设备死了。

协议类型枚举 ScadaProtocolType 用 TINYINT 编码 10/11/20/30/31/40，预留了 5 种以上的扩展空间。这就是为什么不用 SPI——加协议就是加枚举值 + 加 Driver 实现类，IDE 跳转一目了然。

六、调优与踩坑

一是 j2mod 的 TCPMasterConnection 没有显式 isConnected，我用 conn.toString() != null 近似判断，catch Throwable 兜底。这不优雅但能用，商用版应替换成 socket 的 isConnected() && !isClosed()。

二是 OPC UA 的 connectSession().get() 是阻塞的，如果 OPC Server 不响应，借连接会卡 30 秒，连锁阻塞 1Hz 扫表。解决方法是给 Future 加超时：client.connectSession().get(3, TimeUnit.SECONDS)，超时直接抛异常让设备进 RECONNECTING。

三是 PLC4X 0.13 的 S7 连接有个坑，连接串里 rack/slot 必须和 PLC 实际硬件组态一致，错一个就连不上，且报错信息是"timeout"不告诉你哪里错。调试时建议先用 PLC4X 的 cli 工具 plc4x-tools 验证连接串再写代码。

四是 MQTT QoS 设 1 而不是 2。QoS 2 有 PUBREC/PUBREL/PUBCOMP 四次握手，工业场景下位机性能弱，QoS 1 的 at-least-once 已经够，重复消息用点位 ts 去重。

七、商用化检查清单

每种协议的 connParamsJson 字段有没有在 Driver 层强校验？缺字段要给明确报错，不能 NPE。
连接超时有没有配？Modbus、OPC UA 默认超时都很长，必须显式设 3000ms 内。
批量合并的阈值有没有压测过？100 个连续点位应该 1-2 个报文搞定，超过 5 个就要查是不是地址不连续。
Driver 的 isConnected 自检准确吗？池的 validateObject 调它，误判会让好连接被销毁、误判会让坏连接被借出。
反控 writeOne 有没有审计日志？谁在什么时候写了什么值，必须留痕，出了事故能追溯。
OPC UA 安全策略选对了吗？内网可以 None，跨网段必须 Basic256Sha256 + 证书，否则安全审计过不了。

到这里，4 个 Driver 把"怎么连、怎么读、怎么写"讲完了。但"什么时候读"是另一个层面的问题——1Hz 扫表会不会漏？一台设备掉线了怎么自动重连？TDengine 挂了的数据往哪补？这是下一章采集任务调度和断线重连要拆解的。

开篇：为什么是 SCADA 数据采集

万级点位元数据管理