From 4f404a11b015cd63afce8bc6f94fb45a12ead268 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?=E9=82=B9=E8=BF=9B?= <zoujin@qq.com>
Date: Thu, 29 Oct 2026 23:57:54 +0900
Subject: [PATCH] =?UTF-8?q?perf:=20=E8=BF=90=E8=90=A5=E5=95=86=E8=B4=A6?=
 =?UTF-8?q?=E6=88=B7=E5=88=86=E5=8F=91=E6=8C=89=E8=B4=A6=E6=88=B7=E5=B9=B6?=
 =?UTF-8?q?=E8=A1=8C=EF=BC=8C=E4=BA=92=E4=B8=8D=E7=AD=89=E5=BE=85?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

分发器每轮原先逐个处理全部运营商账户，每个账户多次 Redis 往返；
测试环境 Redis 走外网，41 个账户一轮约 19 秒，每个账户每 19 秒只领一次，
慢账户拖住其他账户。改为每轮各账户在独立 goroutine 中抢锁并领取，
一轮耗时取决于最慢账户；账户分发锁不变，多实例下同一账户仍只有一个实例领取。
测试环境 worker 的 Redis 连接池由默认 10 调为 100，避免并行分发与轮询执行争抢连接。

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 docker-compose.prod.yml                     |  2 ++
 internal/polling/carrierqueue/dispatcher.go | 38 +++++++++++++++------
 2 files changed, 29 insertions(+), 11 deletions(-)

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)
 	}
 }