From c2d058bf75c0cea2836782942a206b082a19600f Mon Sep 17 00:00:00 2001 From: lingniu Date: Fri, 3 Jul 2026 08:12:21 +0800 Subject: [PATCH] refactor(go): keep realtime cache vin scoped --- .../production-data-plane-inventory.md | 12 ++++---- docs/architecture/storage-minimal-contract.md | 9 ++++-- docs/go-version-data-flow.html | 14 +++++----- .../internal/realtime/repository.go | 3 ++ .../internal/realtime/repository_test.go | 28 +++++++------------ 5 files changed, 33 insertions(+), 33 deletions(-) diff --git a/docs/architecture/production-data-plane-inventory.md b/docs/architecture/production-data-plane-inventory.md index 2da9adf6..fbe7c106 100644 --- a/docs/architecture/production-data-plane-inventory.md +++ b/docs/architecture/production-data-plane-inventory.md @@ -87,11 +87,11 @@ Redis 使用 DB 50，定位为实时缓存，不作为历史事实来源。 | Key 族 | 用途 | | --- | --- | -| `vehicle:latest:{vehicleKey}` | 跨协议合并后的最新核心字段快照，不重复保存完整 parsed | -| `vehicle:latest:{vehicleKey}:{protocol}` | 单协议最新轻量快照，保留核心 fields 和时间信息，不重复保存完整 parsed | -| `vehicle:realtime-raw:{protocol}:{vehicleKey}` | 单协议最新完整 parsed 状态，是实时完整协议字段的唯一 Redis 副本 | -| `vehicle:online:{vehicleKey}` | 在线状态和 TTL | -| `vehicle:protocols:{vehicleKey}` | 当前车辆最近出现过的协议集合 | +| `vehicle:latest:{vin}` | 跨协议合并后的最新核心字段快照，不重复保存完整 parsed；只写 VIN 非空车辆 | +| `vehicle:latest:{vin}:{protocol}` | 单协议最新轻量快照，保留核心 fields 和时间信息，不重复保存完整 parsed | +| `vehicle:realtime-raw:{protocol}:{vin}` | 单协议最新完整 parsed 状态，是实时完整协议字段的唯一 Redis 副本 | +| `vehicle:online:{vin}` | 在线状态和 TTL | +| `vehicle:protocols:{vin}` | 当前车辆最近出现过的协议集合 | | `vehicle:last_seen` | 最近活跃车辆排序集合 | 审计时活跃 key 族主要是 `vehicle:latest:*`、`vehicle:realtime-raw:*`、`vehicle:online:*`、`vehicle:protocols:*`、`vehicle:last_seen`。旧文档中的 `vehicle:realtime:*`、`vehicle:merged:*` 不是当前活跃 key 族。 @@ -102,4 +102,4 @@ Redis 使用 DB 50，定位为实时缓存，不作为历史事实来源。 2. TDengine 只放高写入时序数据：RAW 证据和位置历史。 3. MySQL 只放低基数业务状态：身份、实时轻量快照、每日里程。 4. Redis 只放当前态，所有 key 都必须允许 TTL 过期后从 Kafka/TDengine/MySQL 重建。 -5. 协议新增字段默认进入 `raw_frames.parsed_json` 和 Redis realtime-raw；只有稳定查询需求出现后才提升为 TDengine/MySQL 列。 +5. 协议新增字段默认进入 `raw_frames.parsed_json`；VIN 非空车辆同步进入 Redis realtime-raw。只有稳定查询需求出现后才提升为 TDengine/MySQL 列。 diff --git a/docs/architecture/storage-minimal-contract.md b/docs/architecture/storage-minimal-contract.md index 2d521410..a2f910de 100644 --- a/docs/architecture/storage-minimal-contract.md +++ b/docs/architecture/storage-minimal-contract.md @@ -26,7 +26,7 @@ | MySQL | `vehicle_daily_mileage` | 日期、VIN、协议、日里程、首末总里程、样本数 | 临时 vehicle key、泛化 metric key/value、自增 id、created_at、每帧细节、位置点列表 | | MySQL | `vehicle_identity_binding` | 人工维护或导入的 VIN、车牌、phone、device_id 映射 | 注册历史、协议状态、自增 id、created_at | | MySQL | `jt808_registration` | JT808 phone 主键下的注册、鉴权、VIN 匹配状态、来源端点 | GB32960/MQTT 注册信息、位置历史、created_at | -| Redis | `vehicle:realtime-raw:{protocol}:{vin}` | 每协议最新完整 parsed 状态 | 历史数据、统计结果 | +| Redis | `vehicle:realtime-raw:{protocol}:{vin}` | 每协议每 VIN 最新完整 parsed 状态 | 无 VIN 临时身份、历史数据、统计结果 | ## 当前应收敛的重复点 @@ -66,7 +66,12 @@ - 写入：车牌从 `vehicle_identity_binding` 按 VIN 主键直查，并使用 VIN 级短 TTL 内存缓存，避免实时帧每条都打 MySQL；缓存不作为事实来源。 - 生产：项目上线前可直接重建这两张 MySQL 表，实时数据会从 Kafka 新消息继续投影。 -7. identity 层只保留两张表。 +7. Redis realtime 只服务已定位 VIN 的正式车辆。 + - 原因：在线状态和实时数据是业务当前态，不是身份排查队列。 + - 无 VIN 的 808 等帧只进入 raw 证据层和注册/绑定排查链路，不写 `vehicle:latest:*`、`vehicle:online:*`、`vehicle:realtime-raw:*`。 + - 绑定补齐 VIN 后，后续新帧自然进入实时态；历史补偿需要从 raw/Kafka 回放。 + +8. identity 层只保留两张表。 - `vehicle_identity_binding`：VIN 与 plate/phone/device_id 的映射，供 808 等协议反查 VIN。 - `jt808_registration`：808 注册、鉴权、最新活跃和 VIN 匹配状态。 - 状态：`vehicle_identity_binding` 使用 VIN 主键；`jt808_registration` 使用 phone 主键；两张表都不保留代理自增主键和 `created_at`。 diff --git a/docs/go-version-data-flow.html b/docs/go-version-data-flow.html index 8224a641..2e414cf7 100644 --- a/docs/go-version-data-flow.html +++ b/docs/go-version-data-flow.html @@ -578,28 +578,28 @@ Redis - vehicle:latest:{vehicle_key} + vehicle:latest:{vin} realtime-api Kafka consumer 跨协议合并后的实时字段快照 查 VIN 是否在线、查 VIN 实时数据 Redis - vehicle:latest:{vehicle_key}:{protocol} + vehicle:latest:{vin}:{protocol} realtime-api Kafka consumer 单协议实时轻量快照 区分 32960/808/MQTT 的实时状态 Redis - vehicle:realtime-raw:{protocol}:{vehicle_key} + vehicle:realtime-raw:{protocol}:{vin} realtime-api Kafka consumer 单协议最新 parsed 全量字段 实时 RAW 字段查看，不走 TDengine 历史扫描 Redis - vehicle:online:{vehicle_key}、vehicle:last_seen + vehicle:online:{vin}、vehicle:last_seen realtime-api Kafka consumer 在线状态、最后接收时间、协议列表 在线判断和近期车辆列表 @@ -639,7 +639,7 @@ VIN 是否在线Redis online keyTTL 内有数据即在线，默认 TTL 600 秒。 VIN 实时数据Redis latest snapshot跨协议合并核心 fields；完整协议字段通过 realtime-raw 查询。 - 单协议实时 RAWRedis realtime-raw查看某 VIN/phone 在某协议下最新 parsed 全量字段。 + 单协议实时 RAWRedis realtime-raw查看某 VIN 在某协议下最新 parsed 全量字段。 @@ -724,8 +724,8 @@ Redis latest snapshot 只保留统一实时核心字段；完整协议 parsed 只放 realtime-raw，避免重复缓存大 JSON。
- 身份解析降级
- VIN 优先，其次 phone、vehicle_key_hint、device_id，保证无 VIN 时仍可临时归档和查询。 + 身份解析边界
+ 实时缓存只写 VIN 非空车辆；无 VIN 帧保留在 RAW 和注册/绑定排查链路，补齐绑定后通过新帧或回放进入实时态。
数据分层查询
diff --git a/go/vehicle-gateway/internal/realtime/repository.go b/go/vehicle-gateway/internal/realtime/repository.go index a61c4924..cc535194 100644 --- a/go/vehicle-gateway/internal/realtime/repository.go +++ b/go/vehicle-gateway/internal/realtime/repository.go @@ -29,6 +29,9 @@ func NewRepository(client *redis.Client, cfg Config) *Repository { func (r *Repository) Update(ctx context.Context, env envelope.FrameEnvelope) error { vin := strings.TrimSpace(env.VIN) + if vin == "" { + return nil + } vehicleKey := strings.TrimSpace(env.VehicleKey()) if vehicleKey == "" || strings.HasSuffix(vehicleKey, ":unknown") { return nil diff --git a/go/vehicle-gateway/internal/realtime/repository_test.go b/go/vehicle-gateway/internal/realtime/repository_test.go index 8943de04..92f44beb 100644 --- a/go/vehicle-gateway/internal/realtime/repository_test.go +++ b/go/vehicle-gateway/internal/realtime/repository_test.go @@ -297,7 +297,7 @@ func TestRepositoryOnlineStatus(t *testing.T) { } } -func TestRepositoryUpdatesPhoneOnlyVehicleKey(t *testing.T) { +func TestRepositorySkipsFramesWithoutVIN(t *testing.T) { repo, closeFn := newTestRepository(t) defer closeFn() ctx := context.Background() @@ -315,23 +315,18 @@ func TestRepositoryUpdatesPhoneOnlyVehicleKey(t *testing.T) { t.Fatalf("Update() error = %v", err) } - vehicleKey := "JT808:13307811170" - merged, err := repo.GetMerged(ctx, vehicleKey) - if err != nil { - t.Fatalf("GetMerged() error = %v", err) + if _, err := repo.GetMerged(ctx, "JT808:13307811170"); err == nil { + t.Fatal("phone-only frame should not create a merged realtime snapshot") } - if merged.VehicleKey != vehicleKey || merged.VIN != "" { - t.Fatalf("unexpected identity: %#v", merged) + if _, err := repo.GetRealtimeRaw(ctx, "JT808:13307811170", envelope.ProtocolJT808); err == nil { + t.Fatal("phone-only frame should not create realtime-raw") } - if merged.Fields[envelope.FieldTotalMileageKM] != 10003.7 { - t.Fatalf("merged fields = %#v", merged.Fields) - } - status, err := repo.IsOnline(ctx, vehicleKey) + status, err := repo.IsOnline(ctx, "JT808:13307811170") if err != nil { t.Fatalf("IsOnline() error = %v", err) } - if !status.Online || status.VehicleKey != vehicleKey { - t.Fatalf("online status = %#v", status) + if status.Online { + t.Fatalf("phone-only frame should not mark realtime online: %#v", status) } } @@ -404,7 +399,7 @@ func TestHandlerReturnsMergedSnapshot(t *testing.T) { } } -func TestHandlerReturnsPhoneOnlyVehicleKeySnapshot(t *testing.T) { +func TestHandlerReturnsNotFoundForPhoneOnlyVehicleKey(t *testing.T) { repo, closeFn := newTestRepository(t) defer closeFn() if err := repo.Update(context.Background(), envelope.FrameEnvelope{ @@ -420,12 +415,9 @@ func TestHandlerReturnsPhoneOnlyVehicleKeySnapshot(t *testing.T) { req := httptest.NewRequest(http.MethodGet, "/api/realtime/vehicles/JT808:13307811170", nil) rec := httptest.NewRecorder() NewHandler(repo).ServeHTTP(rec, req) - if rec.Code != http.StatusOK { + if rec.Code != http.StatusNotFound { t.Fatalf("status = %d body=%s", rec.Code, rec.Body.String()) } - if !stringsContains(rec.Body.String(), `"vehicle_key":"JT808:13307811170"`) { - t.Fatalf("unexpected body: %s", rec.Body.String()) - } } func TestHandlerReturnsRealtimeRaw(t *testing.T) {