From 3a0d9bd840db95ca315fb3eea06f27a21b595465 Mon Sep 17 00:00:00 2001 From: lingniu Date: Fri, 3 Jul 2026 19:45:00 +0800 Subject: [PATCH] feat(go): batch redis fast writer projections --- docs/ops/go-service-observability.md | 2 +- docs/ops/go-vehicle-ingest-memory.md | 4 +- docs/ops/vehicle-ingest-runbook.md | 2 +- .../cmd/nats-fast-writer/main.go | 23 ++++++++ .../cmd/nats-fast-writer/main_test.go | 58 +++++++++++++++++-- .../internal/realtime/repository.go | 43 +++++++++++++- .../internal/realtime/repository_test.go | 51 ++++++++++++++++ 7 files changed, 173 insertions(+), 10 deletions(-) diff --git a/docs/ops/go-service-observability.md b/docs/ops/go-service-observability.md index 3f37ebf0..5904b77f 100644 --- a/docs/ops/go-service-observability.md +++ b/docs/ops/go-service-observability.md @@ -54,7 +54,7 @@ curl -fsS http://127.0.0.1:20200/readyz | `vehicle_fast_writer_nats_consumer_waiting` | Pull requests waiting on the fast writer durable consumer. | | `vehicle_fast_writer_batch_pending_messages` | NATS fast writer messages fetched but not yet written to TDengine/Redis and acked. | | `vehicle_fast_writer_batch_pending_envelopes` | Valid parsed envelopes in the current NATS fast writer batch. | -| `vehicle_fast_writer_stage_duration_ms_histogram` | Fast writer stage duration histogram for TDengine, Redis, and NATS ack. Labels: `subject`, `stage`, `status`. | +| `vehicle_fast_writer_stage_duration_ms_histogram` | Fast writer stage duration histogram for TDengine, Redis, and NATS ack. The Redis stage is one batch pipeline when the repository supports batch updates. Labels: `subject`, `stage`, `status`. | | `vehicle_history_writes_total` | TDengine history writes. Labels: `topic`, `status`. | | `vehicle_history_batch_pending_messages` | Messages fetched by history writer but not yet flushed and committed. | | `vehicle_history_batch_pending_rows` | Parsed raw envelopes waiting in the current TDengine batch. | diff --git a/docs/ops/go-vehicle-ingest-memory.md b/docs/ops/go-vehicle-ingest-memory.md index 1d7e1558..214beac1 100644 --- a/docs/ops/go-vehicle-ingest-memory.md +++ b/docs/ops/go-vehicle-ingest-memory.md @@ -20,6 +20,8 @@ Go 版本车辆数据接入链路已经作为生产主链路运行在 ECS `115.2 5. 解析字段只投影一次，避免 TDengine/Redis/MySQL 各自重复 flatten。 6. 新长期目标：朝 10W 车辆生产稳定运行优化，容量基线见 `docs/ops/100k-capacity-baseline.md`。 +2026-07-03 之后，`nats-fast-writer` 的快速路径是 TDengine 批写 + Redis 批量 pipeline：一个 NATS fetch batch 先批量写 TDengine，再用 `FastUpdateBatch` 把实时 KV、在线 TTL、last_seen 和协议索引合并到一次 Redis pipeline，最后逐条 ack NATS。 + ## 服务和端口 ECS：`115.29.187.205` @@ -431,7 +433,7 @@ vehicle_fast_writer_stage_duration_ms_histogram_sum{subject,stage,status} `stage` 取值包括 `tdengine`、`redis`、`ack`，分别对应快速写 TDengine raw/location、Redis realtime 投影、NATS ack。带帧率压测时如果某个 stage 的 p99 持续上升，可以直接定位 fast path 是存储瓶颈还是 ack/网络瓶颈。 -2026-07-03 后续优化：`nats-fast-writer` 已从“Fetch batch 后逐条写 TDengine”改为“Fetch batch 后先调用 `history.Writer.AppendAllBatch` 批量写 TDengine，再逐条更新 Redis 和 ack”。这样可以减少 TDengine round trip；Redis/ack 仍按单条执行，避免某条实时投影失败时误 ack。 +2026-07-03 后续优化：`nats-fast-writer` 已从“Fetch batch 后逐条写 TDengine/Redis”改为“Fetch batch 后先调用 `history.Writer.AppendAllBatch` 批量写 TDengine，再调用 `realtime.Repository.FastUpdateBatch` 用一次 Redis pipeline 写实时 KV/在线状态/索引，最后逐条 ack NATS”。这样可以同时减少 TDengine 和 Redis round trip；ack 仍按单条执行，避免存储失败时误 ack。 2026-07-03 后续优化：`nats-fast-writer` 的 TDengine SQL 连接池已从固定 `1` 改为可配置： diff --git a/docs/ops/vehicle-ingest-runbook.md b/docs/ops/vehicle-ingest-runbook.md index 93b57ea5..984a045f 100644 --- a/docs/ops/vehicle-ingest-runbook.md +++ b/docs/ops/vehicle-ingest-runbook.md @@ -136,7 +136,7 @@ go run ./cmd/load-sim \ | `vehicle_fast_writer_nats_consumer_pending` | 持续增长且 `> 10000` | fast-writer 消费 NATS 的速度跟不上入口写入速度。 | | `vehicle_fast_writer_batch_pending_messages` | 持续非 0 或 burst 后不回落 | fast-writer 已拉取 NATS 消息但尚未完成 TDengine/Redis 写入和 ack。 | | `vehicle_fast_writer_batch_pending_envelopes` | 持续非 0 或 burst 后不回落 | fast-writer 当前批次已有有效 envelope 在等待落库或 ack。 | -| `vehicle_fast_writer_stage_duration_ms_histogram_bucket` | 某个 stage 的 p99 连续 5 分钟上升 | NATS 快速写链路在 TDengine、Redis 或 NATS ack 某一阶段变慢；TDengine 阶段可小步调整 `FAST_WRITER_TDENGINE_MAX_OPEN_CONNS`，调整后必须观察是否出现选库错误和 ack-pending 增长。 | +| `vehicle_fast_writer_stage_duration_ms_histogram_bucket` | 某个 stage 的 p99 连续 5 分钟上升 | NATS 快速写链路在 TDengine、Redis 或 NATS ack 某一阶段变慢；TDengine 阶段可小步调整 `FAST_WRITER_TDENGINE_MAX_OPEN_CONNS`，Redis 阶段是整批 pipeline 写入耗时，调整后必须观察是否出现选库错误和 ack-pending 增长。 | | `vehicle_realtime_store_update_duration_ms_histogram_bucket{store="redis"}` | p99 连续 5 分钟上升 | Redis 实时投影变慢，会直接影响 realtime consumer 追平能力。 | | `vehicle_realtime_store_update_duration_ms_histogram_bucket{store="mysql"}` | p99 连续 5 分钟上升 | MySQL 当前态/位置投影变慢，需结合 async queue depth 和 dropped 计数判断。 | | `vehicle_stat_write_duration_ms_histogram_bucket` | p99 连续 5 分钟上升 | MySQL 每日里程统计写入变慢，可能导致 stat Kafka lag 增长。 | diff --git a/go/vehicle-gateway/cmd/nats-fast-writer/main.go b/go/vehicle-gateway/cmd/nats-fast-writer/main.go index 2013d9c7..969fb3ba 100644 --- a/go/vehicle-gateway/cmd/nats-fast-writer/main.go +++ b/go/vehicle-gateway/cmd/nats-fast-writer/main.go @@ -194,6 +194,10 @@ type fastUpdater interface { FastUpdate(context.Context, envelope.FrameEnvelope) error } +type fastBatchUpdater interface { + FastUpdateBatch(context.Context, []envelope.FrameEnvelope) error +} + type fastMessage struct { subject string data []byte @@ -290,6 +294,25 @@ func processFastBatch(ctx context.Context, registry *metrics.Registry, appender if err != nil { return fmt.Errorf("tdengine batch append: %w", err) } + if batchUpdater, ok := updater.(fastBatchUpdater); ok { + started = time.Now() + err = batchUpdater.FastUpdateBatch(ctx, envelopes) + recordFastWriterStageDuration(registry, subject, "redis", statusFromError(err), time.Since(started)) + if err != nil { + return fmt.Errorf("redis fast batch update: %w", err) + } + for _, msg := range validMessages { + if msg.ack != nil { + started = time.Now() + err = msg.ack() + recordFastWriterStageDuration(registry, msg.subject, "ack", statusFromError(err), time.Since(started)) + if err != nil { + return fmt.Errorf("nats ack: %w", err) + } + } + } + return nil + } for i, env := range envelopes { msg := validMessages[i] started = time.Now() diff --git a/go/vehicle-gateway/cmd/nats-fast-writer/main_test.go b/go/vehicle-gateway/cmd/nats-fast-writer/main_test.go index 2f67de67..b4e67e67 100644 --- a/go/vehicle-gateway/cmd/nats-fast-writer/main_test.go +++ b/go/vehicle-gateway/cmd/nats-fast-writer/main_test.go @@ -20,7 +20,7 @@ func TestProcessFastMessageWritesTDengineAndRedisBeforeAck(t *testing.T) { t.Fatal(err) } appender := &recordingFastAppender{} - updater := &recordingFastUpdater{} + updater := &recordingFastSingleUpdater{} ackCount := 0 msg := &fastMessage{data: payload, ack: func() error { ackCount++ @@ -95,7 +95,7 @@ func TestProcessFastBatchAppendsTDengineBatchBeforeRedisAndAck(t *testing.T) { t.Fatal(err) } appender := &recordingFastAppender{} - updater := &recordingFastUpdater{} + updater := &recordingFastSingleUpdater{} ackCount := 0 msgs := []*fastMessage{ {subject: "vehicle.raw.go.jt808.v1", data: firstPayload, ack: func() error { ackCount++; return nil }}, @@ -116,6 +116,38 @@ func TestProcessFastBatchAppendsTDengineBatchBeforeRedisAndAck(t *testing.T) { } } +func TestProcessFastBatchUsesRedisBatchUpdaterWhenAvailable(t *testing.T) { + first := envelope.FrameEnvelope{Protocol: envelope.ProtocolJT808, VIN: "VIN001", EventID: "evt-batch-1"} + second := envelope.FrameEnvelope{Protocol: envelope.ProtocolJT808, VIN: "VIN002", EventID: "evt-batch-2"} + firstPayload, err := first.MarshalJSONBytes() + if err != nil { + t.Fatal(err) + } + secondPayload, err := second.MarshalJSONBytes() + if err != nil { + t.Fatal(err) + } + updater := &recordingFastUpdater{} + ackCount := 0 + msgs := []*fastMessage{ + {subject: "vehicle.raw.go.jt808.v1", data: firstPayload, ack: func() error { ackCount++; return nil }}, + {subject: "vehicle.raw.go.jt808.v1", data: secondPayload, ack: func() error { ackCount++; return nil }}, + } + + if err := processFastBatch(context.Background(), nil, &recordingFastAppender{}, updater, msgs); err != nil { + t.Fatalf("processFastBatch() error = %v", err) + } + if updater.batchCount != 1 || updater.batchRows != 2 { + t.Fatalf("FastUpdateBatch count=%d rows=%d, want count=1 rows=2", updater.batchCount, updater.batchRows) + } + if updater.count != 0 { + t.Fatalf("FastUpdate count=%d, want 0 when batch updater is available", updater.count) + } + if ackCount != 2 { + t.Fatalf("ack count=%d, want 2", ackCount) + } +} + func TestProcessFastBatchExposesPendingMetricsDuringAppend(t *testing.T) { first := envelope.FrameEnvelope{Protocol: envelope.ProtocolJT808, VIN: "VIN001", EventID: "evt-6"} second := envelope.FrameEnvelope{Protocol: envelope.ProtocolJT808, VIN: "VIN002", EventID: "evt-7"} @@ -249,11 +281,29 @@ func (a *recordingFastAppender) AppendAllBatch(_ context.Context, envs []envelop } type recordingFastUpdater struct { - count int - err error + count int + batchCount int + batchRows int + err error } func (u *recordingFastUpdater) FastUpdate(context.Context, envelope.FrameEnvelope) error { u.count++ return u.err } + +func (u *recordingFastUpdater) FastUpdateBatch(_ context.Context, envs []envelope.FrameEnvelope) error { + u.batchCount++ + u.batchRows += len(envs) + return u.err +} + +type recordingFastSingleUpdater struct { + count int + err error +} + +func (u *recordingFastSingleUpdater) FastUpdate(context.Context, envelope.FrameEnvelope) error { + u.count++ + return u.err +} diff --git a/go/vehicle-gateway/internal/realtime/repository.go b/go/vehicle-gateway/internal/realtime/repository.go index f1d2ca73..ad491ea5 100644 --- a/go/vehicle-gateway/internal/realtime/repository.go +++ b/go/vehicle-gateway/internal/realtime/repository.go @@ -42,6 +42,36 @@ func (r *Repository) FastUpdate(ctx context.Context, env envelope.FrameEnvelope) return r.setFastProjection(ctx, vehicleKey, vin, env) } +func (r *Repository) FastUpdateBatch(ctx context.Context, envs []envelope.FrameEnvelope) error { + if len(envs) == 0 { + return nil + } + pipe := r.client.Pipeline() + queued := 0 + for _, env := range envs { + if !envelope.IsRealtimeTelemetryFrame(env) { + continue + } + vin := strings.TrimSpace(env.VIN) + if vin == "" { + continue + } + vehicleKey := strings.TrimSpace(env.VehicleKey()) + if vehicleKey == "" || strings.HasSuffix(vehicleKey, ":unknown") { + continue + } + if err := r.queueFastProjection(ctx, pipe, vehicleKey, vin, env); err != nil { + return err + } + queued++ + } + if queued == 0 { + return nil + } + _, err := pipe.Exec(ctx) + return err +} + func (r *Repository) Update(ctx context.Context, env envelope.FrameEnvelope) error { if !envelope.IsRealtimeTelemetryFrame(env) { return nil @@ -373,6 +403,15 @@ func (r *Repository) setKV(ctx context.Context, vin string, env envelope.FrameEn } func (r *Repository) setFastProjection(ctx context.Context, vehicleKey string, vin string, env envelope.FrameEnvelope) error { + pipe := r.client.Pipeline() + if err := r.queueFastProjection(ctx, pipe, vehicleKey, vin, env); err != nil { + return err + } + _, err := pipe.Exec(ctx) + return err +} + +func (r *Repository) queueFastProjection(ctx context.Context, pipe redis.Pipeliner, vehicleKey string, vin string, env envelope.FrameEnvelope) error { values, types := realtimeKVMapsForEnvelope(env) eventTimeMS := eventTimeOrReceivedMS(env) offlineAfterMS := env.ReceivedAtMS + r.cfg.ttl().Milliseconds() @@ -410,7 +449,6 @@ func (r *Repository) setFastProjection(ctx context.Context, vehicleKey string, v "ttl_seconds": strconv.FormatInt(int64(r.cfg.ttl().Seconds()), 10), "source_endpoint": env.SourceEndpoint, } - pipe := r.client.Pipeline() if len(values) > 0 { pipe.HSet(ctx, realtimeKVValuesKey(env.Protocol, vin), values) pipe.HSet(ctx, realtimeKVTypesKey(env.Protocol, vin), types) @@ -420,8 +458,7 @@ func (r *Repository) setFastProjection(ctx context.Context, vehicleKey string, v pipe.HSet(ctx, onlineStateKey(env.Protocol, vin), state) pipe.ZAdd(ctx, "vehicle:last_seen", redis.Z{Score: float64(env.ReceivedAtMS), Member: onlineMember(env.Protocol, vin)}) pipe.SAdd(ctx, realtimeIndexKey(env.Protocol), vin) - _, err = pipe.Exec(ctx) - return err + return nil } func realtimeKVMapsForEnvelope(env envelope.FrameEnvelope) (map[string]any, map[string]any) { diff --git a/go/vehicle-gateway/internal/realtime/repository_test.go b/go/vehicle-gateway/internal/realtime/repository_test.go index 0822d738..f27a0db2 100644 --- a/go/vehicle-gateway/internal/realtime/repository_test.go +++ b/go/vehicle-gateway/internal/realtime/repository_test.go @@ -430,6 +430,57 @@ func TestRepositoryFastUpdateOnlyWritesPermanentKVAndMinuteOnline(t *testing.T) } } +func TestRepositoryFastUpdateBatchWritesMultipleRealtimeProjections(t *testing.T) { + repo, closeFn := newTestRepository(t) + defer closeFn() + ctx := context.Background() + + err := repo.FastUpdateBatch(ctx, []envelope.FrameEnvelope{ + { + Protocol: envelope.ProtocolGB32960, + VIN: "VIN001", + EventTimeMS: 1000, + ReceivedAtMS: 1100, + Parsed: map[string]any{ + "data_units": []any{map[string]any{"type": "0x01", "name": "vehicle", "value": map[string]any{"soc_percent": 88.0}}}, + }, + }, + { + Protocol: envelope.ProtocolJT808, + VIN: "VIN002", + EventTimeMS: 2000, + ReceivedAtMS: 2200, + MessageID: "0x0200", + Parsed: map[string]any{"location": map[string]any{"longitude": 121.1, "latitude": 30.2}}, + Fields: map[string]any{envelope.FieldLongitude: 121.1, envelope.FieldLatitude: 30.2}, + }, + }) + if err != nil { + t.Fatalf("FastUpdateBatch() error = %v", err) + } + + gbValues, err := repo.client.HGetAll(ctx, "vehicle:rt-kv:GB32960:VIN001:values").Result() + if err != nil { + t.Fatalf("gb values HGetAll error = %v", err) + } + if gbValues["gb32960.vehicle.soc_percent"] != "88" { + t.Fatalf("gb values = %#v", gbValues) + } + jtValues, err := repo.client.HGetAll(ctx, "vehicle:rt-kv:JT808:VIN002:values").Result() + if err != nil { + t.Fatalf("jt values HGetAll error = %v", err) + } + if jtValues["jt808.location.longitude"] != "121.1" || jtValues["jt808.location.latitude"] != "30.2" { + t.Fatalf("jt values = %#v", jtValues) + } + if repo.client.ZScore(ctx, "vehicle:last_seen", "GB32960:VIN001").Err() != nil { + t.Fatal("last_seen should contain GB32960 VIN001") + } + if repo.client.ZScore(ctx, "vehicle:last_seen", "JT808:VIN002").Err() != nil { + t.Fatal("last_seen should contain JT808 VIN002") + } +} + func TestRepositoryListsOnlineStatusesAndPipelineSummary(t *testing.T) { repo, closeFn := newTestRepository(t) defer closeFn()