From c36c9fcd84a7b11ac68e274258a8613d5fb21e31 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=82=B9=E8=BF=9B?= Date: Thu, 29 Oct 2026 23:28:28 +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=E8=BD=AE=E8=AF=A2=E5=88=9D=E5=A7=8B=E5=8C=96=E6=8C=89?= =?UTF-8?q?=E6=89=B9=E5=86=99=E5=85=A5=20Redis?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 初始化每批 2000 张卡，原先逐张写入账户队列，每张卡约 4 次 Redis 往返； 测试环境 Redis 走外网，约 750 张/分钟，9.7 万张卡要 2 小时以上，期间未加载的卡不轮询。 新增 Store.UpsertBatch：一次往返读出各卡原账户与优先队列状态，一次往返写入整批； 单卡 Upsert 复用同一逻辑，写入规则不变。批量写入失败时本轮初始化失败、下一轮重试。 Co-Authored-By: Claude Opus 5.5 --- internal/polling/carrier_bridge.go | 22 +++++++- internal/polling/carrierqueue/store.go | 76 ++++++++++++++++++-------- 2 files changed, 73 insertions(+), 25 deletions(-) diff --git a/internal/polling/carrier_bridge.go b/internal/polling/carrier_bridge.go index e4f8d12c..0f99bc66 100644 --- a/internal/polling/carrier_bridge.go +++ b/internal/polling/carrier_bridge.go @@ -92,6 +92,21 @@ func (c *CarrierScheduling) UpsertCard(ctx context.Context, card *model.IotCard, } } +// UpsertCards 批量把卡写入所属账户队列（初始化按批调用）；fromLastCheck 含义同 DuesForCard。 +func (c *CarrierScheduling) UpsertCards(ctx context.Context, cards []*model.IotCard, fromLastCheck bool) error { + if c == nil || c.store == nil || len(cards) == 0 { + return nil + } + items := make([]carrierqueue.UpsertItem, 0, len(cards)) + for _, card := range cards { + if card == nil { + continue + } + items = append(items, carrierqueue.UpsertItem{CardID: card.ID, CarrierID: card.CarrierID, Dues: c.DuesForCard(ctx, card, fromLastCheck)}) + } + return c.store.UpsertBatch(ctx, items) +} + // RemoveCard 把卡移出所属账户队列；carrierID 为 0 时按到期信息中记录的账户移出。 func (c *CarrierScheduling) RemoveCard(ctx context.Context, cardID, carrierID uint) { if c == nil || c.store == nil { @@ -181,10 +196,11 @@ func (i *CarrierInitializer) Run(ctx context.Context) (int64, error) { if tier := i.scheduling.configMgr.TierResolver(); tier != nil { tier.Prefill(ctx, cards) } - for _, card := range cards { - i.scheduling.UpsertCard(ctx, card, true) - written++ + if err := i.scheduling.UpsertCards(ctx, cards, true); err != nil { + i.setProgress(func(p *InitProgress) { p.Status, p.ErrorMessage = "failed", err.Error() }) + return written, err } + written += int64(len(cards)) lastID = cards[len(cards)-1].ID loaded := written i.setProgress(func(p *InitProgress) { p.LoadedCards, p.LastBatchTime = loaded, time.Now() }) diff --git a/internal/polling/carrierqueue/store.go b/internal/polling/carrierqueue/store.go index 8894c333..fa62ebdf 100644 --- a/internal/polling/carrierqueue/store.go +++ b/internal/polling/carrierqueue/store.go @@ -173,27 +173,63 @@ func (s *Store) Remove(ctx context.Context, cardID, carrierID uint) error { // Upsert 写入卡各检查类型到期时间并按最早到期放入账户队列（已在优先队列中的保留在优先队列）。 // 卡所属账户变化时先移出旧账户队列。dues 为空表示该卡没有任何需要执行的检查类型，移出队列。 func (s *Store) Upsert(ctx context.Context, cardID, carrierID uint, dues map[string]time.Time) error { - dueKey := constants.RedisCardDueKey(cardID) - previous, err := s.redis.HGet(ctx, dueKey, fieldCarrier).Uint64() - if err != nil && !stderrors.Is(err, redis.Nil) { + return s.UpsertBatch(ctx, []UpsertItem{{CardID: cardID, CarrierID: carrierID, Dues: dues}}) +} + +// UpsertItem 是批量写入的一张卡：所属账户与各检查类型到期时间（含义同 Upsert）。 +type UpsertItem struct { + CardID uint + CarrierID uint + Dues map[string]time.Time +} + +// UpsertBatch 按 Upsert 的规则批量写入：先一次往返读出各卡原账户与是否在优先队列，再一次往返写入全部卡。 +// 初始化按批调用，避免每张卡多次往返 Redis。 +func (s *Store) UpsertBatch(ctx context.Context, items []UpsertItem) error { + if len(items) == 0 { + return nil + } + read := s.redis.Pipeline() + previousCmds := make([]*redis.StringCmd, len(items)) + priorityCmds := make([]*redis.FloatCmd, len(items)) + for i, item := range items { + previousCmds[i] = read.HGet(ctx, constants.RedisCardDueKey(item.CardID), fieldCarrier) + priorityCmds[i] = read.ZScore(ctx, constants.RedisCarrierPriorityQueueKey(item.CarrierID), strconv.FormatUint(uint64(item.CardID), 10)) + } + if _, err := read.Exec(ctx); err != nil && !stderrors.Is(err, redis.Nil) { return err } - pipe := s.redis.TxPipeline() - if previous != 0 && uint(previous) != carrierID { - pipe.ZRem(ctx, constants.RedisCarrierPriorityQueueKey(uint(previous)), cardID) - pipe.ZRem(ctx, constants.RedisCarrierQueueKey(uint(previous)), cardID) + write := s.redis.Pipeline() + for i, item := range items { + previous, err := previousCmds[i].Uint64() + if err != nil && !stderrors.Is(err, redis.Nil) { + return err + } + inPriority, _ := priorityCmds[i].Result() + queueUpsert(ctx, write, item, uint(previous), inPriority > 0) } - if len(dues) == 0 { - pipe.ZRem(ctx, constants.RedisCarrierPriorityQueueKey(carrierID), cardID) - pipe.ZRem(ctx, constants.RedisCarrierQueueKey(carrierID), cardID) + _, err := write.Exec(ctx) + return err +} + +// queueUpsert 把一张卡的写入命令加入管道：账户变化时移出旧账户队列；dues 为空移出队列； +// 否则写到期 Hash，并按最早到期放入账户队列（已在优先队列中的保留在优先队列）。 +func queueUpsert(ctx context.Context, pipe redis.Pipeliner, item UpsertItem, previous uint, inPriority bool) { + dueKey := constants.RedisCardDueKey(item.CardID) + if previous != 0 && previous != item.CarrierID { + pipe.ZRem(ctx, constants.RedisCarrierPriorityQueueKey(previous), item.CardID) + pipe.ZRem(ctx, constants.RedisCarrierQueueKey(previous), item.CardID) + } + if len(item.Dues) == 0 { + pipe.ZRem(ctx, constants.RedisCarrierPriorityQueueKey(item.CarrierID), item.CardID) + pipe.ZRem(ctx, constants.RedisCarrierQueueKey(item.CarrierID), item.CardID) pipe.Del(ctx, dueKey) - _, err := pipe.Exec(ctx) - return err + return } - values := map[string]any{fieldCarrier: carrierID} + values := map[string]any{fieldCarrier: item.CarrierID} var earliest time.Time for _, checkType := range constants.AllCheckTypes() { - due, ok := dues[checkType] + due, ok := item.Dues[checkType] if !ok { pipe.HDel(ctx, dueKey, fieldDue+checkType) continue @@ -204,15 +240,11 @@ func (s *Store) Upsert(ctx context.Context, cardID, carrierID uint, dues map[str } } pipe.HSet(ctx, dueKey, values) - _, err = pipe.Exec(ctx) - if err != nil { - return err + queueKey := constants.RedisCarrierQueueKey(item.CarrierID) + if inPriority { + queueKey = constants.RedisCarrierPriorityQueueKey(item.CarrierID) } - priorityKey := constants.RedisCarrierPriorityQueueKey(carrierID) - if inPriority, _ := s.redis.ZScore(ctx, priorityKey, strconv.FormatUint(uint64(cardID), 10)).Result(); inPriority > 0 { - return s.redis.ZAdd(ctx, priorityKey, redis.Z{Score: float64(earliest.UnixMilli()), Member: cardID}).Err() - } - return s.redis.ZAdd(ctx, constants.RedisCarrierQueueKey(carrierID), redis.Z{Score: float64(earliest.UnixMilli()), Member: cardID}).Err() + pipe.ZAdd(ctx, queueKey, redis.Z{Score: float64(earliest.UnixMilli()), Member: item.CardID}) } // CardDue 是卡到期信息的解析结果。