diff --git a/docker-compose.prod.yml b/docker-compose.prod.yml index c1b03620..277fd062 100644 --- a/docker-compose.prod.yml +++ b/docker-compose.prod.yml @@ -120,6 +120,8 @@ services: - JUNHONG_REDIS_PORT=16299 - JUNHONG_REDIS_PASSWORD=cpNbWtAaqgo1YJmbMp3h - JUNHONG_REDIS_DB=6 + # worker 并发轮询与按账户并行分发共用连接池,默认 10 不够 + - JUNHONG_REDIS_POOL_SIZE=100 # JWT 配置(必填) - JUNHONG_JWT_SECRET_KEY=dev-secret-key-for-testing-only-32chars! # 客户端配置(可选,默认 true;false=不强制绑定手机号) diff --git a/internal/polling/carrierqueue/dispatcher.go b/internal/polling/carrierqueue/dispatcher.go index 5459de24..cc05a395 100644 --- a/internal/polling/carrierqueue/dispatcher.go +++ b/internal/polling/carrierqueue/dispatcher.go @@ -137,21 +137,37 @@ func (d *Dispatcher) tick(ctx context.Context, interval time.Duration) { if stopResumeOn { d.dispatchDeviceEvals(ctx, lockTTL) } + // 各账户同时分发、互不等待:一轮耗时取决于最慢的账户,而不是全部账户之和。账户分发锁保证同一账户只有一个实例在领。 + var wg sync.WaitGroup for _, carrier := range carriers { if ctx.Err() != nil { - return + break } - ok, err := d.store.redis.SetNX(ctx, constants.RedisCarrierDispatchLockKey(carrier.id), d.instanceID, lockTTL).Result() - if err != nil || !ok { - continue - } - // 停复机优先:先领取停复机指令占用本秒名额,再领取卡任务。 - if stopResumeOn { - d.dispatchStopResume(ctx, carrier.id) - } - if cardsReady { - d.dispatchCarrier(ctx, carrier) + wg.Add(1) + go func(carrier carrierEntry) { + defer wg.Done() + d.dispatchCarrierRound(ctx, carrier, lockTTL, stopResumeOn, cardsReady) + }(carrier) + } + wg.Wait() +} + +// dispatchCarrierRound 是单个账户的一轮分发:抢账户分发锁,先领停复机指令占用本秒名额,再领卡任务。 +func (d *Dispatcher) dispatchCarrierRound(ctx context.Context, carrier carrierEntry, lockTTL time.Duration, stopResumeOn, cardsReady bool) { + defer func() { + if r := recover(); r != nil { + d.logger.Error("账户分发 panic,已恢复", zap.Uint("carrier_id", carrier.id), zap.Any("panic", r)) } + }() + ok, err := d.store.redis.SetNX(ctx, constants.RedisCarrierDispatchLockKey(carrier.id), d.instanceID, lockTTL).Result() + if err != nil || !ok { + return + } + if stopResumeOn { + d.dispatchStopResume(ctx, carrier.id) + } + if cardsReady { + d.dispatchCarrier(ctx, carrier) } }