diff --git a/docs/ops/go-vehicle-ingest-memory.md b/docs/ops/go-vehicle-ingest-memory.md index 214beac1..af533d41 100644 --- a/docs/ops/go-vehicle-ingest-memory.md +++ b/docs/ops/go-vehicle-ingest-memory.md @@ -22,6 +22,8 @@ Go 版本车辆数据接入链路已经作为生产主链路运行在 ECS `115.2 2026-07-03 之后，`nats-fast-writer` 的快速路径是 TDengine 批写 + Redis 批量 pipeline：一个 NATS fetch batch 先批量写 TDengine，再用 `FastUpdateBatch` 把实时 KV、在线 TTL、last_seen 和协议索引合并到一次 Redis pipeline，最后逐条 ack NATS。 +TDengine writer 的 raw/location 子表创建有进程内单飞保护：同一子表 key 并发首次写入时，只有一个 goroutine 执行 `CREATE TABLE IF NOT EXISTS` 和 tag 更新，其他 goroutine 等待结果后继续 INSERT，避免 10W 车辆启动或回放时对 TDengine 形成重复 DDL 风暴。 + ## 服务和端口 ECS：`115.29.187.205` diff --git a/go/vehicle-gateway/internal/history/writer.go b/go/vehicle-gateway/internal/history/writer.go index 7e627c47..cf192e44 100644 --- a/go/vehicle-gateway/internal/history/writer.go +++ b/go/vehicle-gateway/internal/history/writer.go @@ -40,8 +40,45 @@ type payloadChunk struct { } type tableCache struct { - mu sync.Mutex - seen map[string]struct{} + mu sync.Mutex + seen map[string]struct{} + inflight map[string]*tableEnsure +} + +type tableEnsure struct { + done chan struct{} + err error +} + +func (c *tableCache) doOnce(key string, fn func() error) error { + c.mu.Lock() + if _, ok := c.seen[key]; ok { + c.mu.Unlock() + return nil + } + if c.inflight == nil { + c.inflight = map[string]*tableEnsure{} + } + if entry, ok := c.inflight[key]; ok { + c.mu.Unlock() + <-entry.done + return entry.err + } + entry := &tableEnsure{done: make(chan struct{})} + c.inflight[key] = entry + c.mu.Unlock() + + err := fn() + + c.mu.Lock() + if err == nil { + c.seen[key] = struct{}{} + } + entry.err = err + delete(c.inflight, key) + close(entry.done) + c.mu.Unlock() + return err } func NewWriter(exec Execer) *Writer { @@ -230,53 +267,39 @@ VALUES %s`, w.qualify(table), strings.Join(rows, ","))); err != nil { func (w *Writer) ensureRawChild(ctx context.Context, table string, stable string, env envelope.FrameEnvelope) error { key := stable + "." + table - w.cache.mu.Lock() - _, ok := w.cache.seen[key] - w.cache.mu.Unlock() - if ok { - return nil - } - statement := fmt.Sprintf("CREATE TABLE IF NOT EXISTS %s USING %s TAGS ('%s', '%s', '%s', '%s', '%s')", - w.qualify(table), w.qualify(stable), - quote(string(env.Protocol)), - quote(env.VehicleKey()), - quote(env.VIN), - quote(normalizePhoneTag(env.Phone)), - quote(env.DeviceID)) - if _, err := w.exec.ExecContext(ctx, statement); err != nil { - return err - } - if phone := normalizePhoneTag(env.Phone); phone != "" { - if _, err := w.exec.ExecContext(ctx, fmt.Sprintf("ALTER TABLE %s SET TAG phone = '%s'", w.qualify(table), quote(phone))); err != nil { + return w.cache.doOnce(key, func() error { + statement := fmt.Sprintf("CREATE TABLE IF NOT EXISTS %s USING %s TAGS ('%s', '%s', '%s', '%s', '%s')", + w.qualify(table), w.qualify(stable), + quote(string(env.Protocol)), + quote(env.VehicleKey()), + quote(env.VIN), + quote(normalizePhoneTag(env.Phone)), + quote(env.DeviceID)) + if _, err := w.exec.ExecContext(ctx, statement); err != nil { return err } - } - w.cache.mu.Lock() - w.cache.seen[key] = struct{}{} - w.cache.mu.Unlock() - return nil + if phone := normalizePhoneTag(env.Phone); phone != "" { + if _, err := w.exec.ExecContext(ctx, fmt.Sprintf("ALTER TABLE %s SET TAG phone = '%s'", w.qualify(table), quote(phone))); err != nil { + return err + } + } + return nil + }) } func (w *Writer) ensureLocationChild(ctx context.Context, table string, env envelope.FrameEnvelope) error { key := "vehicle_locations." + table - w.cache.mu.Lock() - _, ok := w.cache.seen[key] - w.cache.mu.Unlock() - if ok { + return w.cache.doOnce(key, func() error { + statement := fmt.Sprintf("CREATE TABLE IF NOT EXISTS %s USING %s TAGS ('%s', '%s')", + w.qualify(table), + w.qualify("vehicle_locations"), + quote(string(env.Protocol)), + quote(strings.TrimSpace(env.VIN))) + if _, err := w.exec.ExecContext(ctx, statement); err != nil { + return err + } return nil - } - statement := fmt.Sprintf("CREATE TABLE IF NOT EXISTS %s USING %s TAGS ('%s', '%s')", - w.qualify(table), - w.qualify("vehicle_locations"), - quote(string(env.Protocol)), - quote(strings.TrimSpace(env.VIN))) - if _, err := w.exec.ExecContext(ctx, statement); err != nil { - return err - } - w.cache.mu.Lock() - w.cache.seen[key] = struct{}{} - w.cache.mu.Unlock() - return nil + }) } func rawValues(env envelope.FrameEnvelope, rawHex string, rawText string, parsedFields string) []any { diff --git a/go/vehicle-gateway/internal/history/writer_test.go b/go/vehicle-gateway/internal/history/writer_test.go index 7e8a6e8f..935d676f 100644 --- a/go/vehicle-gateway/internal/history/writer_test.go +++ b/go/vehicle-gateway/internal/history/writer_test.go @@ -4,6 +4,7 @@ import ( "context" "database/sql" "strings" + "sync" "testing" "time" @@ -318,6 +319,47 @@ func TestSchemaStatementsCreateRawFramePayloadChunks(t *testing.T) { } } +func TestWriterEnsuresSameChildTableOnceUnderConcurrency(t *testing.T) { + exec := newBlockingExec("CREATE TABLE IF NOT EXISTS raw_") + writer := NewWriter(exec) + first := sampleEnvelope() + second := sampleEnvelope() + second.Sequence = 2 + second.EventID = "second" + + var wg sync.WaitGroup + errs := make(chan error, 2) + wg.Add(2) + go func() { + defer wg.Done() + errs <- writer.AppendAll(context.Background(), first) + }() + <-exec.blocked + go func() { + defer wg.Done() + errs <- writer.AppendAll(context.Background(), second) + }() + time.Sleep(25 * time.Millisecond) + exec.release <- struct{}{} + wg.Wait() + close(errs) + for err := range errs { + if err != nil { + t.Fatalf("AppendAll() error = %v", err) + } + } + + if got := exec.count("CREATE TABLE IF NOT EXISTS raw_"); got != 1 { + t.Fatalf("raw child CREATE count = %d, want 1; calls=%#v", got, exec.calls()) + } + if got := exec.count("ALTER TABLE raw_"); got != 1 { + t.Fatalf("raw child ALTER count = %d, want 1; calls=%#v", got, exec.calls()) + } + if got := exec.count("INSERT INTO raw_"); got != 2 { + t.Fatalf("raw insert count = %d, want 2; calls=%#v", got, exec.calls()) + } +} + func sampleEnvelope() envelope.FrameEnvelope { return envelope.FrameEnvelope{ Protocol: envelope.ProtocolJT808, @@ -392,3 +434,45 @@ func (e *recordingExec) ExecContext(_ context.Context, query string, args ...any e.calls = append(e.calls, execCall{query: query, args: args}) return nil, nil } + +type blockingExec struct { + mu sync.Mutex + recorded []execCall + blockPattern string + blocked chan struct{} + release chan struct{} + blockOnce sync.Once +} + +func newBlockingExec(blockPattern string) *blockingExec { + return &blockingExec{ + blockPattern: blockPattern, + blocked: make(chan struct{}), + release: make(chan struct{}), + } +} + +func (e *blockingExec) ExecContext(_ context.Context, query string, args ...any) (sql.Result, error) { + e.mu.Lock() + e.recorded = append(e.recorded, execCall{query: query, args: args}) + e.mu.Unlock() + if strings.Contains(query, e.blockPattern) { + e.blockOnce.Do(func() { + close(e.blocked) + <-e.release + }) + } + return nil, nil +} + +func (e *blockingExec) calls() []execCall { + e.mu.Lock() + defer e.mu.Unlock() + out := make([]execCall, len(e.recorded)) + copy(out, e.recorded) + return out +} + +func (e *blockingExec) count(pattern string) int { + return countSQL(e.calls(), pattern) +}