数据流
理解数据在 CoreC 内部的流转路径，是掌握整个系统行为的关键。本章详细拆解从设备寄存器到北向端点的每一个阶段。
全链路总览
┌─────────┐ Read ┌───────────┐ Push ┌─────────┐ Channel ┌───────────────┐
│ Device │ ───────▶ │ Driver │ ───────▶ │ DataBus │ ────────▶ │ processingLoop│
│ (PLC) │ │ (南向) │ │ chan │ │ (规则匹配) │
└─────────┘ └───────────┘ │ cap= │ └───────┬───────┘
onDriverData │ 8192 │ │
TagValue→DataPoint └────┬────┘ │
│ Broadcast Match
▼ │
┌────────────┐ ▼
│ Subscriber │ ┌──────────────┐
│ (WS/API) │ │ Rule Engine │
└────────────┘ └──────┬───────┘
│ Action
▼
┌──────────┐
│Transport │
│ (北向) │
└──────────┘两个核心数据结构
TagValue —— 驱动层的原始读值
TagValue 是驱动从设备读取后返回的原始结构，只包含与设备直接相关的信息：
type TagValue struct {
Tag string // 测点名
Value any // 原始值
Type DataType // 数据类型
Quality Quality // 质量戳 (good/bad/uncertain)
Timestamp time.Time // 读取时刻
Error error // 读取错误（不序列化）
}此时数据还没有路由元信息——不知道来自哪个驱动、属于哪个分组。这是设备协议层的纯粹表达。
DataPoint —— 引擎层的标准数据单元
DataPoint 是在 DataBus 中流转、经规则匹配、最终推送到北向传输的标准单元：
type DataPoint struct {
Driver string // 来源驱动名（富化填充）
Device string // 设备标识
Group string // 点位分组（来自 TagConfig.Group）
Tag string // 测点名
Value any // 值
Type DataType // 数据类型
Quality Quality // 质量戳
Timestamp time.Time // 时间戳
Metadata map[string]string // 扩展元数据
}富化过程（TagValue → DataPoint）
富化发生在引擎的 onDriverData 回调中。调度器每次读取完成后，将 []TagValue 交给引擎，引擎逐个补充路由元信息（驱动名与分组）：
func (e *CoreCEngine) onDriverData(driver string, values []core.TagValue) {
// 按 driver 取出 tag → group 映射，一次读取避免逐点加锁
groups := e.tagGroups[driver]
for _, v := range values {
point := core.DataPoint{
Driver: driver, // ← 富化：补充驱动名
Tag: v.Tag,
Value: v.Value,
Type: v.Type,
Quality: v.Quality,
Timestamp: v.Timestamp,
}
if groups != nil {
point.Group = groups[v.Tag] // ← 富化：按测点名查表补充分组
}
e.dataBus.Push(point) // 推入数据总线
}
}为什么需要富化？
规则引擎需要依据 driver、group、tag 等字段做路由决策，北向传输的主题模板（如 factory/{{.Driver}}/{{.Group}}/{{.Tag}}）也需要这些字段。富化将设备层的原始读值升级为携带完整路由上下文的标准数据点。
管道各阶段详解
阶段 1：调度触发（Scheduler）
调度器为每个「驱动 + 采集周期」组合创建一个独立 goroutine，使用 time.Ticker 周期触发：
驱动 demo-plc 有三个点位：
temperature interval=1s
pressure interval=1s
pump_status interval=2s
调度器按 interval 分组，创建两个任务：
任务 demo-plc_1s → [temperature, pressure] 每 1s 触发
任务 demo-plc_2s → [pump_status] 每 2s 触发每个任务一个 goroutine，互不阻塞。触发时调用 driver.Read(ctx, tags) 批量读取该周期内所有点位。
按周期分组的优势
同一周期的点位合并为一次 Read 调用，减少协议交互次数；驱动内部还会把连续地址合并为批量请求（Modbus 每次最多 125 个寄存器）。例如 100 个 1s 点位每秒只需一次 Read 调用，而非 100 次调度。
阶段 2：驱动读取（Driver.Read）
驱动将测点名翻译为协议地址，发起实际读取。以 Modbus 为例：
Read(ctx, ["temperature", "pressure"])
→ 解析 address "40001" → 保持寄存器 0
→ client.ReadFloat32(0)
→ 返回 TagValue{Tag:"temperature", Value:42.5, Quality:good, ...}读取失败时，对应点位的 Quality 设为 bad 并填充 Error，但不会中断整批读取——其他点位仍正常返回。
阶段 3：富化与入队（onDriverData → DataBus.Push）
如上文所述，TagValue 被富化为 DataPoint 后推入 DataBus。
阶段 4：数据总线（DataBus）
DataBus 是连接采集协程与处理协程的 Go channel 组，主通道容量 8192（global.engine.data-bus-size 可调）：
e.dataBus = NewDataBus(8192) // engine/lifecycle_support.go除主通道外，DataBus 还带一条高优先级通道（容量为主通道的 1/4，上限 1024）：采集周期 ≤ 1s 的快周期任务走该通道，由专属 worker 消费，避免低频批量读取的突发流量挤压高频采集。
丢弃策略（Drop-Oldest）
当总线满时，Push 采用 drop-oldest 策略而非阻塞：
func (b *DataBus) Push(point core.DataPoint) {
select {
case b.ch <- point: // 正常入队
default: // 队列满
select {
case <-b.ch: // 丢弃最旧的一个，计数 +1
b.pushDropped.Add(1)
default:
}
select { // 再次尝试非阻塞入队；若槽位被并发抢占则丢弃新数据
case b.ch <- point:
default:
b.pushDropped.Add(1)
}
}
}为什么丢弃而非阻塞？
工业场景中过时数据的价值快速衰减。如果选择阻塞采集协程，会导致：
- 调度周期被拖长，引发 overrun 雪崩
- 设备连接因超时断开
- 所有点位采集集体停滞
丢弃最旧数据保证采集协程始终以设备节奏运行，过载通过 PushDropped() 计数器暴露给运维。
总线满时丢弃的最旧数据通过 DataBus.PushDropped()（主通道）与 HighPriorityPushDropped()（高优先级通道）计数；下游慢订阅者被跳过的数据通过 DataBus.Dropped() 计数。三者再加上日志丢弃，聚合到引擎统计的 TotalDropped 字段，可通过 /stats API 查询。
阶段 5：处理循环（processingLoop）
引擎启动多个 worker goroutine（默认 runtime.NumCPU() 个，global.engine.workers 可调）并行消费 DataBus 主通道；另有默认 2 个高优先级 worker（global.engine.high-priority-workers 可调）只消费高优先级通道。每个 worker 对每个 DataPoint 依次执行：
func (e *CoreCEngine) processingLoop() {
for {
select {
case <-e.ctx.Done():
return
case point, ok := <-e.dataBus.Channel():
// ① 更新最新值缓存
e.cache.Update(point)
// ② 广播给订阅者（WebSocket / API 流）
e.dataBus.Broadcast(point)
// ③ 规则匹配
result := e.ruleEngine.Match(point)
if result == nil {
continue // 无规则匹配，数据止于此
}
// ④ 执行动作
switch result.Rule.Action() {
case core.ActionDrop:
continue
case core.ActionAlert:
e.fireAlert(point, result.Rule)
e.publishToTargets(point, result.Targets)
case core.ActionTransform:
e.publishToTargets(e.applyTransform(point, result.Transform), result.Targets)
case core.ActionForward, core.ActionMirror:
e.publishToTargets(point, result.Targets)
}
}
}
}注：实际
processingLoop在读取DataPoint后会最先应用坏质量策略（global.engine.on-bad-quality：publish（默认）/drop/mark-and-publish/alert），再依次更新缓存、广播、规则匹配与执行动作。上方代码片段为简化展示，省略了坏质量策略步骤。
阶段 6：最新值缓存（LatestCache）
每个 DataPoint 都会更新 LatestCache。为降低多 worker 并发更新时的锁竞争，缓存采用 64 分片（shard）设计——按驱动名哈希取模分散到 64 个分片，每个分片各自持有 RWMutex：
const numCacheShards = 64
type LatestCache struct {
shards [numCacheShards]cacheShard // 按 driver 名 FNV-1a 哈希取模定位分片
}
type cacheShard struct {
mu sync.RWMutex
drivers map[string]*driverCache // 该分片内的 driver → tag → DataPoint
}缓存支撑按需查询：
GET /drivers/{name}/tags—— 返回某驱动所有点位的最新值GET /tags—— 返回全部驱动全部点位的最新值，支持?driver=、?tag=（测点名子串）、?prefix=（测点名前缀）三个可选查询参数做服务端过滤
按驱动筛选优先使用
/drivers/{name}/tags路径参数形式；/tags?driver=<name>是等价的查询参数写法。
缓存与调度的关系
LatestCache 保存每个点位的最近一次读值，无论该值是否被规则转发，也无论质量好坏（仅 on-bad-quality: drop 策略的点不进缓存）。即使某点位被规则 drop，缓存中仍有值可供 API 查询。这使得"采集"与"转发"两个关注点彻底解耦。
阶段 7：广播给订阅者（Broadcast）
DataBus.Broadcast 将数据点推送给所有通过 Subscribe(filter) 注册的订阅者（主要是 WebSocket /tags/stream 端点）：
func (b *DataBus) Broadcast(point core.DataPoint) {
for _, sub := range b.subscribers {
select {
case sub.ch <- point: // 正常推送
default: // 订阅者缓冲满
b.dropped.Add(1) // 跳过并计数
}
}
}每个订阅者有独立的 256 容量缓冲 channel。慢订阅者（如 WebSocket 客户端处理不过来）会被跳过而非阻塞主流水线，跳过次数计入 DataBus.Dropped()。
阶段 8：规则匹配与动作执行
详见 规则引擎 章节。处理循环将匹配结果中的目标传输列表交给 publishToTargets：
func (e *CoreCEngine) publishToTargets(point core.DataPoint, targets []string) {
// ① 锁内快照：查找 transport 和 batcher，避免锁外访问 map
e.mu.RLock()
snapshot := make([]publishTargetEntry, 0, len(targets))
for _, targetName := range targets {
transport, ok := e.transports[targetName]
if !ok { continue } // 目标传输不存在，告警并跳过
entry := publishTargetEntry{name: targetName, transport: transport}
if batcher, ok := e.batchers[targetName]; ok {
entry.batcher = batcher // 有批量配置则走 batcher
}
snapshot = append(snapshot, entry)
}
e.mu.RUnlock()
// ② 锁外发布：避免慢网络 I/O 阻塞管理操作（AddDriver/RemoveDriver 等）
for _, entry := range snapshot {
if entry.batcher != nil {
entry.batcher.publish(e.ctx, point) // 进入批量缓冲
continue
}
entry.transport.Publish(e.ctx, point) // 直接发布
}
}锁分离设计
publishToTargets 采用锁内快照 + 锁外 I/O 模式：在 e.mu.RLock() 下仅做 map 查找并快照到切片，释放读锁后再调用 Publish/batcher.publish。这避免了慢速网络发布阻塞 AddDriver/RemoveDriver/AddTransport/RemoveTransport 等需要写锁的管理操作。
阶段 9：北向发布（Transport.Publish）
传输将 DataPoint 序列化后发往外部系统。以 MQTT 为例，使用主题模板渲染目标主题，JSON 序列化后发布并等待完成令牌（publish-timeout，默认 5s）：
DataPoint{Driver:"demo-plc", Group:"sensors", Tag:"temperature", Value:42.5}
→ 主题模板 "factory/{{.Driver}}/{{.Group}}/{{.Tag}}"
→ 渲染为 "factory/demo-plc/sensors/temperature"
→ JSON 序列化
→ paho MQTT 发布 (QoS 1，等待 token 完成或超时)反向数据流：控制指令下发
除了上行采集，CoreC 还支持下行控制。北向传输可以通过 OnCommand() 暴露一个 <-chan WriteCommand 反向通道：
MQTT Command Topic ──订阅──▶ Transport.OnCommand() ──chan──▶ startCommandListener ──▶ Driver.Write ──▶ 设备引擎为每个传输在 AddTransport 时启动一个独立的 startCommandListener goroutine 监听该传输的 OnCommand() 通道，收到 WriteCommand 后调用对应驱动的 Write 方法写入设备：
type WriteCommand struct {
Driver string // 目标驱动
Device string // 设备标识
Tag string // 目标测点
Value any // 待写入值
Type DataType // 值类型
}以 MQTT 为例，配置 command-topic: "factory/commands/#"（须同时配置 command-secret，消息需携带 HMAC-SHA256 签名，见传输）后，向该主题发布 JSON 格式的 WriteCommand 即可触发设备写入，形成闭环控制。
数据质量戳
CoreC 借鉴 OPC UA 的质量戳概念，每个数据点都携带 Quality 字段：
| 质量 | 值 | 含义 |
|---|---|---|
good | 0 | 读取成功，值可信 |
bad | 1 | 读取失败（连接断开、地址错误等） |
uncertain | 2 | 不确定（预留扩展） |
质量为 bad 的数据点在默认策略（on-bad-quality: publish）下仍会进入总线与缓存（便于感知设备状态），规则可以据此过滤：
rules:
- name: drop-bad-quality
match: "quality == 'bad'"
action: drop
priority: 1引擎级还有独立的坏质量策略 global.engine.on-bad-quality（publish / drop / mark-and-publish / alert），在处理循环最前端生效，先于缓存与规则；drop 策略的丢弃量计入 corec_bad_quality_dropped_total 指标。
性能特征
| 环节 | 并发模型 | 阻塞风险 |
|---|---|---|
| 调度读取 | 每任务一 goroutine，互不阻塞 | 单任务超时不影响其他任务 |
| 数据总线 | Go channel，无锁 | 满时 drop-oldest，从不阻塞采集 |
| 处理循环 | 多 goroutine 并行消费（默认 runtime.NumCPU() 个 worker + 2 个高优先级 worker） | 规则匹配为纯内存操作，极快 |
| 北向发布 | 同步 Publish，逐目标串行 | 发布失败计数，不阻塞主流水线 |
| 订阅广播 | 非阻塞 select，慢消费者跳过 | 永不阻塞主流水线 |
吞吐量参考
引擎内置基准测试（engine/bench_test.go）覆盖总线推送、广播、缓存与规则匹配等环节。瓶颈通常在北向传输的远端系统（MQTT Broker、HTTP 服务端）而非核心本身。通过 mirror 动作分摊到多传输、或调整 batch-size / flush-interval 可进一步优化。