From 079bd76d3833e312452d04b6f39a8922ec811ca1 Mon Sep 17 00:00:00 2001 From: lingniu Date: Thu, 2 Jul 2026 14:22:11 +0800 Subject: [PATCH] chore(go): tune gateway async publish defaults --- go/vehicle-gateway/cmd/gateway/main.go | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/go/vehicle-gateway/cmd/gateway/main.go b/go/vehicle-gateway/cmd/gateway/main.go index 3aed504c..8f87f485 100644 --- a/go/vehicle-gateway/cmd/gateway/main.go +++ b/go/vehicle-gateway/cmd/gateway/main.go @@ -169,7 +169,7 @@ func buildSink(ctx context.Context, logger *slog.Logger) (eventbus.Sink, error) }) spoolDir := strings.TrimSpace(os.Getenv("KAFKA_SPOOL_DIR")) if spoolDir != "" { - replayBatchSize := envInt("KAFKA_SPOOL_REPLAY_BATCH_SIZE", 1000) + replayBatchSize := envInt("KAFKA_SPOOL_REPLAY_BATCH_SIZE", 200) durable := eventbus.NewDurableSink(out, eventbus.DurableConfig{Directory: spoolDir, ReplayBatchSize: replayBatchSize}) interval := time.Duration(envInt("KAFKA_SPOOL_REPLAY_INTERVAL_MS", 1000)) * time.Millisecond go durable.ReplayLoop(ctx, interval, func(err error) { @@ -179,8 +179,8 @@ func buildSink(ctx context.Context, logger *slog.Logger) (eventbus.Sink, error) out = durable } if envBool("KAFKA_ASYNC_ENABLED", true) { - queueSize := envInt("KAFKA_ASYNC_QUEUE_SIZE", 10000) - workers := envInt("KAFKA_ASYNC_WORKERS", 1) + queueSize := envInt("KAFKA_ASYNC_QUEUE_SIZE", 100000) + workers := envInt("KAFKA_ASYNC_WORKERS", 8) timeout := time.Duration(envInt("KAFKA_ASYNC_PUBLISH_TIMEOUT_MS", 30000)) * time.Millisecond out = eventbus.NewAsyncSink(out, eventbus.AsyncConfig{ QueueSize: queueSize,