shishengliang/lingniu-vehicle-ingest
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

chore(go): tune gateway async publish defaults

Browse Source
This commit is contained in:
lingniu
2026-07-02 14:22:11 +08:00
parent 5145d156bc
commit 079bd76d38
1 changed files with 3 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
go/vehicle-gateway/cmd/gateway/main.go
View File

@@ -169,7 +169,7 @@ func buildSink(ctx context.Context, logger *slog.Logger) (eventbus.Sink, error)
}) })
spoolDir := strings.TrimSpace(os.Getenv("KAFKA_SPOOL_DIR")) spoolDir := strings.TrimSpace(os.Getenv("KAFKA_SPOOL_DIR"))
if spoolDir != "" { 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}) durable := eventbus.NewDurableSink(out, eventbus.DurableConfig{Directory: spoolDir, ReplayBatchSize: replayBatchSize})
interval := time.Duration(envInt("KAFKA_SPOOL_REPLAY_INTERVAL_MS", 1000)) * time.Millisecond interval := time.Duration(envInt("KAFKA_SPOOL_REPLAY_INTERVAL_MS", 1000)) * time.Millisecond
go durable.ReplayLoop(ctx, interval, func(err error) { 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 out = durable
} }
if envBool("KAFKA_ASYNC_ENABLED", true) { if envBool("KAFKA_ASYNC_ENABLED", true) {
queueSize := envInt("KAFKA_ASYNC_QUEUE_SIZE", 10000) queueSize := envInt("KAFKA_ASYNC_QUEUE_SIZE", 100000)
workers := envInt("KAFKA_ASYNC_WORKERS", 1) workers := envInt("KAFKA_ASYNC_WORKERS", 8)
timeout := time.Duration(envInt("KAFKA_ASYNC_PUBLISH_TIMEOUT_MS", 30000)) * time.Millisecond timeout := time.Duration(envInt("KAFKA_ASYNC_PUBLISH_TIMEOUT_MS", 30000)) * time.Millisecond
out = eventbus.NewAsyncSink(out, eventbus.AsyncConfig{ out = eventbus.NewAsyncSink(out, eventbus.AsyncConfig{
QueueSize: queueSize, QueueSize: queueSize,