背景
在车联网场景中，数采平台需要实时接收数百至上千辆车辆的上报数据，涵盖实时遥测、心跳、故障、状态变更等多种消息类型。每辆车按 10~30 秒间隔上报，千辆车并发意味着每秒需要处理数十到上百条 MQTT 消息，且每条消息需经过解码、去重、存储、规则评估等多个环节。
本文结合一个 Go 语言实现的车辆数据采集平台，分享 MQTT 消息处理链路中的高并发设计实践。
整体架构
数据从车载终端到最终落库的链路如下：
车载终端 → EMQX → collector(订阅/解码/分发) → [state, storage, rule]核心设计原则：
- 入口简单：设备只感知 EMQX，不感知内部架构
- 处理解耦：collector 不直接写业务逻辑，通过标准事件模型分发给下游模块
- 可水平扩展：EMQX 共享订阅支持多 collector 实例并行消费
关键设计
1. 共享订阅实现多实例负载均衡
EMQX 的共享订阅（Shared Subscription）是实现多 collector 实例并行消费的核心机制。订阅时在 topic 前加上$share/<group>/前缀：
// mqtt_client.gofunc(p*pahoClient)Subscribe(topics[]string,qosbyte,handlerfunc(RawMessage))error{for_,topic:=rangetopics{subTopic:=topicifp.sharedGroup!=""{subTopic=fmt.Sprintf("$share/%s/%s",p.sharedGroup,topic)}// 订阅 subTopic，如 $share/tbox-collect/atlas/+/realtimedata}}效果：同一个 topic 下的消息会被 EMQX 轮询分配到同组内的不同 collector 实例，实现无状态水平扩展。当前配置的共享组名为tbox-collect，后续新增实例无需修改配置。
2. Redis Stream 持久化队列：避免内存消息丢失
问题
纯内存通道（chan RawMessage）在进程重启时会丢失未处理的消息。对于车辆遥测数据，丢失意味着历史数据出现缺口。
方案
引入 Redis Stream 作为持久化缓冲层。MQTT 回调不再直接投递到内存 channel，而是通过XADD写入 Redis Stream：
// collector.gofunc(c*Collector)onMessage(msg RawMessage){ifc.streamIngress!=nil{// 路径 A：Redis Stream 持久化c.streamIngress.Enqueue(ctx,msg.Topic,msg.Payload,msg.QoS,msg.ReceivedAt)return}// 路径 B：内存 channel（开发/轻量环境）select{casec.msgCh<-msg:default:c.log.Warn("collector queue full, dropping message")}}消费端使用单协程XREAD顺序消费，保证严格 FIFO，处理成功后XDEL删除：
// ingress.go - 核心消费循环func(i*Ingress)RunSequential(ctx context.Context,processfunc(context.Context,Delivery)error){afterID:="0-0"for{streams,_:=i.rdb.XRead(ctx,&redis.XReadArgs{Streams:[]string{i.stream,afterID},Count:1,Block:i.cfg.Block,}).Result()// 处理每条消息，成功后推进游标for_,msg:=rangest.Messages{iferr:=process(ctx,d);err!=nil{continue// 失败不推进，重启后重新处理}afterID=msg.ID}}}关键细节：
- 遥测数据（高频）攒批写入 ClickHouse 后批量
XDEL，避免逐条删除的性能开销 - 故障、心跳等低频数据在同步落库成功后立即
XDEL - 处理失败的消息不会被删除，进程重启后自动重新消费
- Stream 设置
MAXLEN ~10000近似裁剪，防止内存无限增长
3. 统一事件模型与分发器设计
标准事件类型
所有 MQTT 消息解码后转换为统一的标准事件：
| 消息类型 | 事件结构 | 下游分发目标 |
|---|---|---|
realtimedata | VehicleTelemetryEvent | state + storage + rule |
truckstatus | VehicleStatusEvent | state + rule |
heatbeat | HeartbeatEvent | state + storage |
fault_event | FaultEvent | state + storage + rule |
event | RawEvent | storage + rule |
分发器接口
通过接口解耦 collector 与下游模块：
typeEventDispatcherinterface{Dispatch(ctx context.Context,eventinterface{})errorDispatchTelemetry(ctx context.Context,event*TelemetryDispatch)errorDispatchFault(ctx context.Context,event*FaultDispatch)error// ...}ModuleDispatcher根据事件类型调用对应的下游处理器（StateHandler / StorageHandler / RuleHandler），每个处理器的错误独立记录，互不影响。
4. Protobuf / JSON 双模式解码
车端协议存在 Protobuf 和 JSON 两种编码格式。解码策略：
- 优先尝试 Protobuf 解码（通过注册表查找对应消息类型的解码器）
- Protobuf 失败时降级为 JSON 解码（配置项
json_fallback: true） - 解码失败记录到 ClickHouse 的
dwd_vehicle_decode_error表，方便回溯排查
func(d*StandardDecoder)Decode(ctx context.Context,meta EventMeta,payload[]byte)(interface{},error){ifd.cfg.ProtoEnabled{ifdecoder,ok:=d.protoDecoders[meta.MessageType];ok{event,err:=decoder.Decode(ctx,meta,payload)iferr==nil{returnevent,nil}if!d.cfg.JSONFallback{returnnil,err}}}returnd.jsonDecoder.Decode(ctx,meta,payload)}5. 消息去重：基于时间窗口的滑动去重
车辆在弱网环境下可能重发消息。去重逻辑基于VIN + MessageType + Payload哈希生成去重键，在一个可配置的时间窗口内（默认 5 秒）判定重复：
func(c*Collector)processIncoming(ctx context.Context,msg RawMessage)error{ifc.config.DedupEnabled{key:=buildDedupKey(meta,msg.Payload)dup,_:=c.deduper.IsDuplicate(ctx,key,c.config.DedupWindow)ifdup{returnc.deleteStreamEntries(ctx,msg.RedisStreamID)}}// ... 解码和分发}去重器通过接口抽象，支持内存实现（MemoryDeduper，适合小规模）和 Redis 实现（适合多实例共享状态）。
6. 遥测攒批写入与异步 Flush
遥测数据是最高频的消息类型。每条实时写入 ClickHouse 会产生大量小 batch，吞吐低下。方案：
- 内存缓冲区攒批，达到阈值（默认 40 条）或空闲超时（默认 10 秒）时触发异步 flush
- flush 时浅拷贝缓冲区，在后台协程执行批量写入，不阻塞采集链路
- 写入失败时将数据重新放回缓冲区，配合重试机制（指数退避，最多 3 次）
func(s*StorageWriter)WriteTelemetry(ctx context.Context,telemetry*models.VehicleTelemetryEvent,streamIDstring)error{s.mu.Lock()s.telemetryBuffer=append(s.telemetryBuffer,bufferedTelemetry{ev:telemetry,streamID:streamID})should:=s.shouldFlush()// 条件：buffer >= 40 或空闲 >= 10ss.mu.Unlock()ifshould{s.flushTelemetryAsync()// 异步 flush，不阻塞调用方}returnnil}7. 动态 Topic 订阅管理
生产环境中需要支持新增或停用消息类型，但不应重启服务。平台通过数据库管理订阅配置，collector 定时刷新（默认 60 秒）：
func(c*Collector)refreshRoutes(ctx context.Context){routes,_:=c.config.RouteStore.LoadEnabled(ctx)added,removed:=c.config.RouteRegistry.Refresh(routes)// 动态取消已停用的 topiciflen(removed)>0{c.mqtt.Unsubscribe(removed)}// 动态订阅新增的 topiciflen(added)>0{c.mqtt.Subscribe(added,c.config.QoS,c.onMessage)}}降级策略：数据库不可用时依次回退到注册表缓存 → 配置文件 → 硬编码默认值。
可观测性
全链路通过 Prometheus 指标监控：
| 指标 | 用途 |
|---|---|
tbox_collect_mqtt_messages_received_total | 按 topic 统计消息接收量 |
tbox_collect_mqtt_decode_duration_seconds | 解码耗时分布（直方图） |
tbox_collect_mqtt_dispatch_duration_seconds | 分发耗时分布 |
tbox_collect_mqtt_queue_length | 内存队列积压深度 |
tbox_collect_state_vehicles_online | 在线车辆数 |
tbox_collect_mqtt_redis_stream_enqueue_total | Redis Stream 入队次数 |
配合 Grafana 面板，可以实时观察消息处理瓶颈、队列积压和车辆在线状态。
经验总结
| 决策 | 理由 |
|---|---|
| Redis Stream 而非纯内存队列 | 进程重启不丢数据，成本远低于引入 Kafka |
| 单协程 XREAD 顺序消费 | 避免并发消费带来的顺序性问题，当前规模下吞吐足够 |
| 遥测攒批 40 条异步 flush | 平衡实时性与写入吞吐，ClickHouse 批量写入性能远优于逐条 |
| Protobuf 优先 + JSON 降级 | 兼顾新终端（Protobuf）和老终端（JSON），平滑迁移 |
| 接口抽象（Decoder / Deduper / EventDispatcher） | 为未来扩展预留空间（如 Kafka 替换 Redis Stream） |
后续演进
当前架构支撑千辆级规模。当车辆规模和消息量持续增长时，计划在 collector 与下游模块之间引入 Kafka 事件总线：
EMQX → collector → Kafka → [storage, rule, state]得益于统一事件模型和接口抽象，这一演进只需替换消息分发层，核心处理逻辑无需修改。
