c36c9fcd84
All checks were successful
构建并部署到测试环境（无 SSH） / build-and-deploy (push) Successful in 11m6s
初始化每批 2000 张卡，原先逐张写入账户队列，每张卡约 4 次 Redis 往返； 测试环境 Redis 走外网，约 750 张/分钟，9.7 万张卡要 2 小时以上，期间未加载的卡不轮询。 新增 Store.UpsertBatch：一次往返读出各卡原账户与优先队列状态，一次往返写入整批； 单卡 Upsert 复用同一逻辑，写入规则不变。批量写入失败时本轮初始化失败、下一轮重试。 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
317 lines
13 KiB
Go
317 lines
13 KiB
Go
package carrierqueue
|
||
|
||
import (
|
||
"context"
|
||
stderrors "errors"
|
||
"fmt"
|
||
"strconv"
|
||
"strings"
|
||
"time"
|
||
|
||
"github.com/redis/go-redis/v9"
|
||
|
||
"github.com/break/junhong_cmp_fiber/pkg/constants"
|
||
)
|
||
|
||
// Store 封装按运营商账户调度的 Redis 结构与 Lua 原子操作。
|
||
type Store struct {
|
||
redis *redis.Client
|
||
}
|
||
|
||
// NewStore 创建账户队列存储。
|
||
func NewStore(redisClient *redis.Client) *Store {
|
||
return &Store{redis: redisClient}
|
||
}
|
||
|
||
// Client 返回底层 Redis 客户端。
|
||
func (s *Store) Client() *redis.Client {
|
||
return s.redis
|
||
}
|
||
|
||
// ClaimInput 是一次账户领取的参数。
|
||
type ClaimInput struct {
|
||
CarrierID uint
|
||
Lease time.Duration
|
||
InflightLimit int
|
||
PriorityRatio float64 // 0 表示不限
|
||
TokenPrefix string // 本轮派发 token 前缀（实例 + 轮次），token = 前缀:card_id
|
||
GroupBudget map[string]int // 限流组本秒剩余；nil 表示不按 QPS 限制
|
||
TypeGroup map[string]string // 检查类型 → 限流组
|
||
CountRealname bool // 预计调用数是否计入实名（账户未开实名管控时不计）
|
||
}
|
||
|
||
// Claimed 是一张被领取的卡。
|
||
type Claimed struct {
|
||
CardID uint
|
||
Token string
|
||
FromPriority bool
|
||
}
|
||
|
||
// Claim 执行一次账户领取，返回领取到的卡与本轮清理的过期执行数（处理中崩溃或超时次数）。
|
||
func (s *Store) Claim(ctx context.Context, input ClaimInput) ([]Claimed, int64, error) {
|
||
budget := ""
|
||
if input.GroupBudget != nil {
|
||
parts := make([]string, 0, len(input.GroupBudget))
|
||
for group, remaining := range input.GroupBudget {
|
||
parts = append(parts, group+"="+strconv.Itoa(remaining))
|
||
}
|
||
budget = strings.Join(parts, ";")
|
||
if budget == "" {
|
||
budget = "none=0"
|
||
}
|
||
}
|
||
typeParts := make([]string, 0, len(input.TypeGroup))
|
||
for checkType, group := range input.TypeGroup {
|
||
typeParts = append(typeParts, checkType+"="+group)
|
||
}
|
||
countRealname := "0"
|
||
if input.CountRealname {
|
||
countRealname = "1"
|
||
}
|
||
raw, err := claimScript.Run(ctx, s.redis,
|
||
[]string{constants.RedisCarrierPriorityQueueKey(input.CarrierID), constants.RedisCarrierQueueKey(input.CarrierID), constants.RedisCarrierRunningKey(input.CarrierID)},
|
||
time.Now().UnixMilli(), input.Lease.Milliseconds(), input.InflightLimit, strconv.FormatFloat(input.PriorityRatio, 'f', 3, 64),
|
||
input.TokenPrefix, budget, strings.Join(typeParts, ";"), countRealname,
|
||
).Slice()
|
||
if err != nil {
|
||
return nil, 0, err
|
||
}
|
||
if len(raw) == 0 {
|
||
return nil, 0, nil
|
||
}
|
||
expired, _ := raw[0].(int64)
|
||
claimed := make([]Claimed, 0, (len(raw)-1)/3)
|
||
for index := 1; index+2 < len(raw); index += 3 {
|
||
cardID, _ := strconv.ParseUint(fmt.Sprint(raw[index]), 10, 64)
|
||
claimed = append(claimed, Claimed{CardID: uint(cardID), Token: fmt.Sprint(raw[index+1]), FromPriority: fmt.Sprint(raw[index+2]) == "1"})
|
||
}
|
||
return claimed, expired, nil
|
||
}
|
||
|
||
// AcquireExec 抢执行权：返回 1 抢到，0 处理中标记已不是该 token（过期或重复投递），-1 该 token 已有执行者。
|
||
func (s *Store) AcquireExec(ctx context.Context, cardID uint, token, execValue string, ttl time.Duration) (int64, error) {
|
||
return acquireExecScript.Run(ctx, s.redis, []string{constants.RedisCardClaimKey(cardID), constants.RedisCardExecKey(cardID)},
|
||
token, execValue, ttl.Milliseconds()).Int64()
|
||
}
|
||
|
||
// Renew 续期租约与执行权；返回 false 表示 token 或执行权已不一致，调用方必须停止后续步骤。
|
||
func (s *Store) Renew(ctx context.Context, cardID, carrierID uint, token, execValue string, lease, execTTL time.Duration) (bool, error) {
|
||
result, err := renewScript.Run(ctx, s.redis,
|
||
[]string{constants.RedisCardClaimKey(cardID), constants.RedisCardExecKey(cardID), constants.RedisCarrierRunningKey(carrierID),
|
||
constants.RedisCarrierPriorityQueueKey(carrierID), constants.RedisCarrierQueueKey(carrierID)},
|
||
token, execValue, lease.Milliseconds(), execTTL.Milliseconds(), time.Now().UnixMilli(), cardID).Int64()
|
||
return result == 1, err
|
||
}
|
||
|
||
// Finish 收尾：写回业务结果（set 字段与删除字段），token 一致时释放处理中标记并按最早到期设置 score。
|
||
// 返回 false 表示 token 已不一致，只写入了业务结果。
|
||
func (s *Store) Finish(ctx context.Context, cardID, carrierID uint, token string, set map[string]string, del []string) (bool, error) {
|
||
args := []any{token, cardID, time.Now().UnixMilli(), len(set)}
|
||
for field, value := range set {
|
||
args = append(args, field, value)
|
||
}
|
||
for _, field := range del {
|
||
args = append(args, field)
|
||
}
|
||
result, err := finishScript.Run(ctx, s.redis,
|
||
[]string{constants.RedisCardDueKey(cardID), constants.RedisCardClaimKey(cardID), constants.RedisCardExecKey(cardID),
|
||
constants.RedisCarrierRunningKey(carrierID), constants.RedisCarrierPriorityQueueKey(carrierID), constants.RedisCarrierQueueKey(carrierID)},
|
||
args...).Int64()
|
||
return result == 1, err
|
||
}
|
||
|
||
// Release 发布失败时按 token 删除处理中标记并把 score 恢复为当前时间。
|
||
func (s *Store) Release(ctx context.Context, cardID, carrierID uint, token string) error {
|
||
return releaseScript.Run(ctx, s.redis,
|
||
[]string{constants.RedisCardClaimKey(cardID), constants.RedisCarrierRunningKey(carrierID),
|
||
constants.RedisCarrierPriorityQueueKey(carrierID), constants.RedisCarrierQueueKey(carrierID)},
|
||
token, cardID, time.Now().UnixMilli()).Err()
|
||
}
|
||
|
||
// CheckRequest 是一次请求检查。
|
||
type CheckRequest struct {
|
||
CardID uint
|
||
CarrierID uint
|
||
CheckTypes []string
|
||
Cooldown time.Duration
|
||
ConfirmDelays []time.Duration
|
||
Expected string // 延迟确认的期望结果（card_status 为 online / offline），满足时取消后续确认
|
||
// Normal 为 true 时不放入优先队列：只把到期设为现在、在普通队列中立即到期（手动触发走普通通道）。
|
||
Normal bool
|
||
}
|
||
|
||
// RequestCheck 把卡的指定检查类型设为立即到期放入优先队列；返回实际请求的类型数，卡不在轮询中返回 -1。
|
||
func (s *Store) RequestCheck(ctx context.Context, request CheckRequest) (int64, error) {
|
||
if len(request.CheckTypes) == 0 {
|
||
return 0, nil
|
||
}
|
||
confirm := make([]string, 0, len(request.ConfirmDelays))
|
||
for _, delay := range request.ConfirmDelays {
|
||
confirm = append(confirm, strconv.FormatInt(int64(delay/time.Second), 10))
|
||
}
|
||
toPriority := "1"
|
||
if request.Normal {
|
||
toPriority = "0"
|
||
}
|
||
args := []any{time.Now().UnixMilli(), request.CardID, request.Cooldown.Milliseconds(), strings.Join(confirm, ","), request.Expected, toPriority}
|
||
for _, checkType := range request.CheckTypes {
|
||
args = append(args, checkType)
|
||
}
|
||
return requestCheckScript.Run(ctx, s.redis,
|
||
[]string{constants.RedisCardDueKey(request.CardID), constants.RedisCarrierPriorityQueueKey(request.CarrierID),
|
||
constants.RedisCarrierQueueKey(request.CarrierID), constants.RedisCardClaimKey(request.CardID)},
|
||
args...).Int64()
|
||
}
|
||
|
||
// Remove 把卡移出所属账户队列并删除到期信息（关闭轮询、删除卡）。
|
||
func (s *Store) Remove(ctx context.Context, cardID, carrierID uint) error {
|
||
return removeScript.Run(ctx, s.redis,
|
||
[]string{constants.RedisCarrierPriorityQueueKey(carrierID), constants.RedisCarrierQueueKey(carrierID), constants.RedisCardDueKey(cardID)},
|
||
cardID).Err()
|
||
}
|
||
|
||
// Upsert 写入卡各检查类型到期时间并按最早到期放入账户队列（已在优先队列中的保留在优先队列）。
|
||
// 卡所属账户变化时先移出旧账户队列。dues 为空表示该卡没有任何需要执行的检查类型，移出队列。
|
||
func (s *Store) Upsert(ctx context.Context, cardID, carrierID uint, dues map[string]time.Time) error {
|
||
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
|
||
}
|
||
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)
|
||
}
|
||
_, 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)
|
||
return
|
||
}
|
||
values := map[string]any{fieldCarrier: item.CarrierID}
|
||
var earliest time.Time
|
||
for _, checkType := range constants.AllCheckTypes() {
|
||
due, ok := item.Dues[checkType]
|
||
if !ok {
|
||
pipe.HDel(ctx, dueKey, fieldDue+checkType)
|
||
continue
|
||
}
|
||
values[fieldDue+checkType] = due.UnixMilli()
|
||
if earliest.IsZero() || due.Before(earliest) {
|
||
earliest = due
|
||
}
|
||
}
|
||
pipe.HSet(ctx, dueKey, values)
|
||
queueKey := constants.RedisCarrierQueueKey(item.CarrierID)
|
||
if inPriority {
|
||
queueKey = constants.RedisCarrierPriorityQueueKey(item.CarrierID)
|
||
}
|
||
pipe.ZAdd(ctx, queueKey, redis.Z{Score: float64(earliest.UnixMilli()), Member: item.CardID})
|
||
}
|
||
|
||
// CardDue 是卡到期信息的解析结果。
|
||
type CardDue struct {
|
||
CarrierID uint
|
||
Priority bool
|
||
Types map[string]TypeDue
|
||
}
|
||
|
||
// TypeDue 是卡某检查类型的到期与失败信息。
|
||
type TypeDue struct {
|
||
Due time.Time
|
||
HasDue bool
|
||
LastOK time.Time
|
||
FailCount int
|
||
FirstFailed time.Time
|
||
LastError string
|
||
Confirm []time.Duration
|
||
Expected string
|
||
}
|
||
|
||
// ReadDue 读取并解析卡到期信息；Hash 不存在时返回空结构。
|
||
func (s *Store) ReadDue(ctx context.Context, cardID uint) (CardDue, error) {
|
||
raw, err := s.redis.HGetAll(ctx, constants.RedisCardDueKey(cardID)).Result()
|
||
if err != nil {
|
||
return CardDue{}, err
|
||
}
|
||
return ParseDue(raw), nil
|
||
}
|
||
|
||
// ParseDue 解析卡到期 Hash。
|
||
func ParseDue(raw map[string]string) CardDue {
|
||
result := CardDue{Types: map[string]TypeDue{}}
|
||
if value, err := strconv.ParseUint(raw[fieldCarrier], 10, 64); err == nil {
|
||
result.CarrierID = uint(value)
|
||
}
|
||
result.Priority = raw[fieldPriority] == "1"
|
||
for _, checkType := range constants.AllCheckTypes() {
|
||
item := TypeDue{}
|
||
if ms, err := strconv.ParseInt(raw[fieldDue+checkType], 10, 64); err == nil {
|
||
item.Due, item.HasDue = time.UnixMilli(ms), true
|
||
}
|
||
if ms, err := strconv.ParseInt(raw[fieldOK+checkType], 10, 64); err == nil && ms > 0 {
|
||
item.LastOK = time.UnixMilli(ms)
|
||
}
|
||
item.FailCount, _ = strconv.Atoi(raw[fieldFail+checkType])
|
||
if ms, err := strconv.ParseInt(raw[fieldFirstFail+checkType], 10, 64); err == nil && ms > 0 {
|
||
item.FirstFailed = time.UnixMilli(ms)
|
||
}
|
||
item.LastError = raw[fieldError+checkType]
|
||
for _, part := range strings.Split(raw[fieldConfirm+checkType], ",") {
|
||
if seconds, err := strconv.Atoi(strings.TrimSpace(part)); err == nil && seconds > 0 {
|
||
item.Confirm = append(item.Confirm, time.Duration(seconds)*time.Second)
|
||
}
|
||
}
|
||
item.Expected = raw[fieldExpected+checkType]
|
||
result.Types[checkType] = item
|
||
}
|
||
return result
|
||
}
|
||
|
||
// Field 名称导出给执行器组装收尾写入。
|
||
func FieldDue(checkType string) string { return fieldDue + checkType }
|
||
func FieldOK(checkType string) string { return fieldOK + checkType }
|
||
func FieldFail(checkType string) string { return fieldFail + checkType }
|
||
func FieldFirstFail(checkType string) string { return fieldFirstFail + checkType }
|
||
func FieldError(checkType string) string { return fieldError + checkType }
|
||
func FieldConfirm(checkType string) string { return fieldConfirm + checkType }
|
||
func FieldExpected(checkType string) string { return fieldExpected + checkType }
|