DolphinDB 流表只能追加，如何维护每台设备的实时最新状态？

2026-11-04数据小陈

可以用 heartbeatStream 接收心跳事件，通过流订阅将数据写入 latestKeyedTable。这张表以 device_id 为主键、以 ts 判断新旧，每台设备只保留一条记录。看板直接查询状态表，无需每秒扫描心跳历史。

为什么选择 latestKeyedTable？

普通 keyedTable 写入相同主键时会更新已有记录，适合维护状态缓存。但它按写入顺序覆盖，延迟到达的旧心跳可能覆盖较新的状态。参见官方 keyedTable 文档。

latestKeyedTable 会比较时间列：同一主键的新记录时间戳小于已有记录时，不会覆盖已有记录。这个接口在官方 3.00.0 文档中已经存在，可用于 3.00.4 场景。参见 latestKeyedTable 文档。

数据流可以组织为：

设备心跳 → heartbeatStream → 流订阅 → deviceLatestState → 运维看板

建立状态表和订阅

假设现有 heartbeatStream 的列顺序为 ts、device_id、status、temperature、battery_pct，类型分别为 TIMESTAMP、SYMBOL、SYMBOL、DOUBLE、DOUBLE。如果实际定义不同，需要相应调整目标表。

share latestKeyedTable(
    `device_id,
    `ts,
    100:0,
    `ts`device_id`status`temperature`battery_pct,
    [TIMESTAMP, SYMBOL, SYMBOL, DOUBLE, DOUBLE]
) as deviceLatestState

subscribeTable(
    tableName="heartbeatStream",
    actionName="maintainDeviceLatestState",
    offset=0,
    handler=tableInsert{deviceLatestState},
    msgAsTable=true,
    batchSize=0
)

订阅收到数据后，会将记录插入状态表，由 latestKeyedTable 执行主键去重和时间戳比较。batchSize=0 表示消息到达后尽快处理。由于订阅是异步的，看板仍可能短暂落后于入口流表。参数语义见官方 subscribeTable 文档。

这里的 offset=0 表示从流表创建时的第一条记录开始消费，前提是所需数据仍可读取。如果旧数据已被清理，需要选择有效的起始位置，再通过状态快照或历史数据补齐初始状态。

用新心跳推进状态

设备发生故障时，向流表追加一条时间更新的完整记录：

insert into heartbeatStream values(
    2024.03.01T09:00:30.000,
    `AC_B1_01,
    `fault,
    24.0,
    97.5
)

订阅处理完成后，看板即可查询状态表：

select
    device_id,
    status,
    temperature,
    battery_pct,
    ts as last_ts
from deviceLatestState

查询单台设备时，可以使用主键条件：

select *
from deviceLatestState
where device_id = `AC_B1_01

如果只有 20 个不同的 device_id，状态表就只维护这 20 个键对应的记录。新增设备会增加行数，退役设备需要按业务规则清理。看板全量刷新时读取的是设备状态集合，查询规模不再随心跳历史累计增长。

流表的增长仍需单独控制

状态表解决了看板查询问题，入口流表的缓存仍需配置。100:0、10000:0 中的容量是初始内存分配量，不是行数上限。

生产环境可在配置流表持久化时，通过 enableTableShareAndPersistence 设置 cacheSize，控制流表保留在内存中的记录数量，并结合磁盘日志保留策略安排回放。该函数用于空流表的共享和持久化初始化，现有流表应按实际部署状态调整。参见官方 enableTableShareAndPersistence 文档。

状态表属于内存状态，重启后需要从其他来源恢复。可以加载状态快照，再衔接后续心跳；也可以从仍保留的事件中重建。如果某台设备的最后一次心跳早已超出日志保留范围，仅回放近期日志无法恢复它的状态。

容易混淆的几个细节

latestKeyedStreamTable 不会把流表压缩成每台设备一行。它会比较时间戳，将符合条件的新记录继续追加。官方示例中，同一主键的旧行和更新后的行会同时存在，因此不能直接替代这里的状态表。参见 latestKeyedStreamTable 文档。
最大时间戳和最后到达的记录不是同一个概念。原查询不能仅凭 last(status) 与 max(ts) 就认定各字段来自同一条最新事件；乱序和相同时间戳都需要明确规则。状态表方案直接维护整条记录，避免分别拼接字段。
时间戳需要可信。示例按设备上报的 ts 判断新旧。对于设备时钟异常或相同时间戳下的冲突，需要约定校时、事件序号或冲突处理规则。
设备停止发送心跳后，不会自动变成 offline。可以根据最后接收时间与当前时间的差值判断超时，阈值应结合 30 秒心跳间隔和网络抖动确定。若要准确表示通信存活情况，建议另存服务端接收时间。

按 365 天计算，20 台设备每 30 秒各发送一条心跳，一年约产生 20 × 365 × 24 × 60 × 2 = 21,024,000 条记录。应避免反复扫描这些历史记录，但不能仅凭这个数量断言服务器会过载，实际表现还取决于硬件、缓存、查询和并发情况。