a4e3fe4d8f
All checks were successful
构建并部署到测试环境（无 SSH） / build-and-deploy (push) Successful in 9m24s
- 开关开启但分发器心跳缺失时回退为当场执行，避免指令无人消费 - 待确认指令先查上游状态再判断；执行期间被覆盖时把待确认带到新指令 - 同方向失败退避中重复写入不提前执行时间 - 领取容量把执行上限与本秒剩余 QPS 分开计算 - 队列复机共用单卡复机锁；执行上下文取消后按结果未知确认 - 消费者并行停止；读取最近调用时间失败时不覆盖；成功但终结日志失败按成功处理 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
355 lines
14 KiB
Go
355 lines
14 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"
|
||
)
|
||
|
||
// StopResumeCommand 是写入停复机队列的一条指令。
|
||
type StopResumeCommand struct {
|
||
CardID uint
|
||
CarrierID uint
|
||
Action string // constants.StopResumeActionStop / StopResumeActionStart
|
||
Reason string // 停机原因
|
||
Scenes []string // 触发场景
|
||
Source string // 请求来源（日志与排查）
|
||
}
|
||
|
||
// CommandState 是卡当前停复机指令的解析结果。
|
||
type CommandState struct {
|
||
Exists bool
|
||
Action string
|
||
Reason string
|
||
Scenes []string
|
||
Source string
|
||
CarrierID uint
|
||
Version int64
|
||
FailCount int
|
||
FirstFailed time.Time
|
||
LastError string
|
||
Pending bool
|
||
Unknown int
|
||
WrittenAt time.Time
|
||
NextAt time.Time // 队列中的可执行时间（零值表示不在队列中）
|
||
LastCallAt time.Time
|
||
}
|
||
|
||
// WriteCommand 写入或覆盖卡的停复机指令，返回新版本号。
|
||
func (s *Store) WriteCommand(ctx context.Context, cmd StopResumeCommand) (int64, error) {
|
||
if cmd.CardID == 0 || cmd.CarrierID == 0 {
|
||
return 0, stderrors.New("停复机指令缺少卡或运营商账户")
|
||
}
|
||
return srWriteScript.Run(ctx, s.redis,
|
||
[]string{constants.RedisCardStopResumeCmdKey(cmd.CardID), constants.RedisCarrierStopResumeQueueKey(cmd.CarrierID),
|
||
constants.RedisCardStopResumeClaimKey(cmd.CardID)},
|
||
cmd.Action, cmd.Reason, constants.JoinStopResumeScenes(cmd.Scenes), cmd.Source, cmd.CarrierID, time.Now().UnixMilli(), cmd.CardID,
|
||
).Int64()
|
||
}
|
||
|
||
// SRClaimInput 是一次账户停复机领取的参数。
|
||
type SRClaimInput struct {
|
||
CarrierID uint
|
||
Lease time.Duration
|
||
InflightLimit int // 账户停复机执行上限（含执行中）
|
||
Budget int // 停复机所在限流组本秒剩余；-1 表示不按 QPS 限制
|
||
TokenPrefix string
|
||
MinInterval time.Duration
|
||
}
|
||
|
||
// SRClaimed 是一条被领取的停复机指令。
|
||
type SRClaimed struct {
|
||
CardID uint
|
||
Token string
|
||
Version int64
|
||
Action string
|
||
Pending bool
|
||
}
|
||
|
||
// ClaimStopResume 领取账户到期停复机指令，返回领取结果与本轮清理的过期执行数。
|
||
func (s *Store) ClaimStopResume(ctx context.Context, input SRClaimInput) ([]SRClaimed, int64, error) {
|
||
raw, err := srClaimScript.Run(ctx, s.redis,
|
||
[]string{constants.RedisCarrierStopResumeQueueKey(input.CarrierID), constants.RedisCarrierStopResumeRunningKey(input.CarrierID)},
|
||
time.Now().UnixMilli(), input.Lease.Milliseconds(), input.InflightLimit, input.TokenPrefix, input.MinInterval.Milliseconds(), input.Budget,
|
||
).Slice()
|
||
if err != nil || len(raw) == 0 {
|
||
return nil, 0, err
|
||
}
|
||
expired, _ := raw[0].(int64)
|
||
claimed := make([]SRClaimed, 0, (len(raw)-1)/5)
|
||
for index := 1; index+4 < len(raw); index += 5 {
|
||
cardID, _ := strconv.ParseUint(fmt.Sprint(raw[index]), 10, 64)
|
||
version, _ := strconv.ParseInt(fmt.Sprint(raw[index+2]), 10, 64)
|
||
claimed = append(claimed, SRClaimed{CardID: uint(cardID), Token: fmt.Sprint(raw[index+1]), Version: version,
|
||
Action: fmt.Sprint(raw[index+3]), Pending: fmt.Sprint(raw[index+4]) == "1"})
|
||
}
|
||
return claimed, expired, nil
|
||
}
|
||
|
||
// AcquireStopResumeExec 抢停复机指令执行权：1 抢到，0 处理中标记已不是该 token，-1 已有执行者。
|
||
func (s *Store) AcquireStopResumeExec(ctx context.Context, cardID uint, token, execValue string, ttl time.Duration) (int64, error) {
|
||
return acquireExecScript.Run(ctx, s.redis, []string{constants.RedisCardStopResumeClaimKey(cardID), constants.RedisCardStopResumeExecKey(cardID)},
|
||
token, execValue, ttl.Milliseconds()).Int64()
|
||
}
|
||
|
||
// SRFinish 是一次停复机指令收尾。
|
||
type SRFinish struct {
|
||
CardID uint
|
||
CarrierID uint
|
||
Token string
|
||
Version int64
|
||
Done bool // true 结束并删除指令
|
||
Unknown bool // 本次结果未知：版本已变大时把待确认带到新指令
|
||
NextAt time.Time
|
||
Set map[string]string
|
||
Del []string
|
||
}
|
||
|
||
// FinishStopResume 停复机指令收尾：返回 0 token 不一致、1 正常、2 版本已变大保留新指令。
|
||
func (s *Store) FinishStopResume(ctx context.Context, finish SRFinish) (int64, error) {
|
||
done := "0"
|
||
if finish.Done {
|
||
done = "1"
|
||
}
|
||
now := time.Now()
|
||
next := finish.NextAt
|
||
if next.IsZero() || next.Before(now) {
|
||
next = now
|
||
}
|
||
unknown := "0"
|
||
if finish.Unknown {
|
||
unknown = "1"
|
||
}
|
||
args := []any{finish.Token, finish.CardID, finish.Version, done, next.UnixMilli(), now.UnixMilli(), unknown, len(finish.Set)}
|
||
for field, value := range finish.Set {
|
||
args = append(args, field, value)
|
||
}
|
||
for _, field := range finish.Del {
|
||
args = append(args, field)
|
||
}
|
||
return srFinishScript.Run(ctx, s.redis,
|
||
[]string{constants.RedisCardStopResumeCmdKey(finish.CardID), constants.RedisCardStopResumeClaimKey(finish.CardID),
|
||
constants.RedisCardStopResumeExecKey(finish.CardID), constants.RedisCarrierStopResumeRunningKey(finish.CarrierID),
|
||
constants.RedisCarrierStopResumeQueueKey(finish.CarrierID)},
|
||
args...).Int64()
|
||
}
|
||
|
||
// ReleaseStopResume 发布失败时释放停复机指令处理中标记并恢复为立即可领取。
|
||
func (s *Store) ReleaseStopResume(ctx context.Context, cardID, carrierID uint, token string) error {
|
||
return srReleaseScript.Run(ctx, s.redis,
|
||
[]string{constants.RedisCardStopResumeClaimKey(cardID), constants.RedisCarrierStopResumeRunningKey(carrierID),
|
||
constants.RedisCarrierStopResumeQueueKey(carrierID)},
|
||
token, cardID, time.Now().UnixMilli()).Err()
|
||
}
|
||
|
||
// MarkStopResumeCall 记录本卡一次停复机网关调用（停机与复机合计），供单卡最小重试间隔判断。
|
||
func (s *Store) MarkStopResumeCall(ctx context.Context, cardID uint, at time.Time, keep time.Duration) error {
|
||
if keep < time.Hour {
|
||
keep = time.Hour
|
||
}
|
||
return s.redis.Set(ctx, constants.RedisCardStopResumeLastCallKey(cardID), at.UnixMilli(), keep).Err()
|
||
}
|
||
|
||
// LastStopResumeCall 返回本卡最近一次停复机网关调用时间；没有记录返回零值。
|
||
func (s *Store) LastStopResumeCall(ctx context.Context, cardID uint) (time.Time, error) {
|
||
ms, err := s.redis.Get(ctx, constants.RedisCardStopResumeLastCallKey(cardID)).Int64()
|
||
if stderrors.Is(err, redis.Nil) {
|
||
return time.Time{}, nil
|
||
}
|
||
if err != nil {
|
||
return time.Time{}, err
|
||
}
|
||
return time.UnixMilli(ms), nil
|
||
}
|
||
|
||
// ReadCommand 读取卡当前停复机指令与队列位置。
|
||
func (s *Store) ReadCommand(ctx context.Context, cardID uint) (CommandState, error) {
|
||
raw, err := s.redis.HGetAll(ctx, constants.RedisCardStopResumeCmdKey(cardID)).Result()
|
||
if err != nil {
|
||
return CommandState{}, err
|
||
}
|
||
state := ParseCommand(raw)
|
||
if last, lastErr := s.LastStopResumeCall(ctx, cardID); lastErr == nil {
|
||
state.LastCallAt = last
|
||
}
|
||
if state.Exists && state.CarrierID > 0 {
|
||
member := strconv.FormatUint(uint64(cardID), 10)
|
||
if score, scoreErr := s.redis.ZScore(ctx, constants.RedisCarrierStopResumeQueueKey(state.CarrierID), member).Result(); scoreErr == nil {
|
||
state.NextAt = time.UnixMilli(int64(score))
|
||
}
|
||
}
|
||
return state, nil
|
||
}
|
||
|
||
// ParseCommand 解析停复机指令 Hash。
|
||
func ParseCommand(raw map[string]string) CommandState {
|
||
state := CommandState{Exists: raw[srFieldAction] != ""}
|
||
state.Action, state.Reason, state.Source, state.LastError = raw[srFieldAction], raw[srFieldReason], raw[srFieldSource], raw[srFieldError]
|
||
state.Scenes = constants.SplitStopResumeScenes(raw[srFieldScenes])
|
||
if value, err := strconv.ParseUint(raw[srFieldCarrier], 10, 64); err == nil {
|
||
state.CarrierID = uint(value)
|
||
}
|
||
state.Version, _ = strconv.ParseInt(raw[srFieldVersion], 10, 64)
|
||
state.FailCount, _ = strconv.Atoi(raw[srFieldFail])
|
||
state.Unknown, _ = strconv.Atoi(raw[srFieldUnknown])
|
||
state.Pending = raw[srFieldPending] == "1"
|
||
if ms, err := strconv.ParseInt(raw[srFieldFirst], 10, 64); err == nil && ms > 0 {
|
||
state.FirstFailed = time.UnixMilli(ms)
|
||
}
|
||
if ms, err := strconv.ParseInt(raw[srFieldAt], 10, 64); err == nil && ms > 0 {
|
||
state.WrittenAt = time.UnixMilli(ms)
|
||
}
|
||
return state
|
||
}
|
||
|
||
// 停复机指令字段名导出给执行器组装收尾写入。
|
||
const (
|
||
SRFieldFail = srFieldFail
|
||
SRFieldFirst = srFieldFirst
|
||
SRFieldError = srFieldError
|
||
SRFieldPending = srFieldPending
|
||
SRFieldUnknown = srFieldUnknown
|
||
)
|
||
|
||
// RequestDeviceEval 请求一次设备维度判断：递增版本、合并场景、设备进入判断队列。
|
||
func (s *Store) RequestDeviceEval(ctx context.Context, deviceID uint, scenes []string) error {
|
||
if deviceID == 0 {
|
||
return stderrors.New("设备判断缺少设备")
|
||
}
|
||
args := []any{deviceID, time.Now().UnixMilli()}
|
||
for _, scene := range constants.SplitStopResumeScenes(constants.JoinStopResumeScenes(scenes)) {
|
||
args = append(args, scene)
|
||
}
|
||
return deviceRequestScript.Run(ctx, s.redis,
|
||
[]string{constants.RedisDeviceEvalVersionKey(deviceID), constants.RedisDeviceEvalScenesKey(deviceID), constants.RedisDeviceEvalQueueKey,
|
||
constants.RedisDeviceEvalClaimKey(deviceID)},
|
||
args...).Err()
|
||
}
|
||
|
||
// DeviceEvalClaimed 是一次被领取的设备判断。
|
||
type DeviceEvalClaimed struct {
|
||
DeviceID uint
|
||
Token string
|
||
Version string
|
||
Scenes []string
|
||
}
|
||
|
||
// ClaimDeviceEvals 领取到期设备判断。
|
||
func (s *Store) ClaimDeviceEvals(ctx context.Context, lease time.Duration, limit int, tokenPrefix string) ([]DeviceEvalClaimed, int64, error) {
|
||
raw, err := deviceClaimScript.Run(ctx, s.redis, []string{constants.RedisDeviceEvalQueueKey, constants.RedisDeviceEvalRunningKey},
|
||
time.Now().UnixMilli(), lease.Milliseconds(), limit, tokenPrefix).Slice()
|
||
if err != nil || len(raw) == 0 {
|
||
return nil, 0, err
|
||
}
|
||
expired, _ := raw[0].(int64)
|
||
claimed := make([]DeviceEvalClaimed, 0, (len(raw)-1)/4)
|
||
for index := 1; index+3 < len(raw); index += 4 {
|
||
deviceID, _ := strconv.ParseUint(fmt.Sprint(raw[index]), 10, 64)
|
||
claimed = append(claimed, DeviceEvalClaimed{DeviceID: uint(deviceID), Token: fmt.Sprint(raw[index+1]),
|
||
Version: fmt.Sprint(raw[index+2]), Scenes: constants.SplitStopResumeScenes(fmt.Sprint(raw[index+3]))})
|
||
}
|
||
return claimed, expired, nil
|
||
}
|
||
|
||
// AcquireDeviceEvalLock 抢设备判断锁（执行权）：1 抢到，0 处理中标记已不是该 token，-1 已有判断在执行。
|
||
func (s *Store) AcquireDeviceEvalLock(ctx context.Context, deviceID uint, token, value string, ttl time.Duration) (int64, error) {
|
||
return acquireExecScript.Run(ctx, s.redis, []string{constants.RedisDeviceEvalClaimKey(deviceID), constants.RedisDeviceEvalLockKey(deviceID)},
|
||
token, value, ttl.Milliseconds()).Int64()
|
||
}
|
||
|
||
// FinishDeviceEval 设备判断收尾：返回 0 token 不一致、1 完成、2 版本变大已重新排队、3 已推后。
|
||
func (s *Store) FinishDeviceEval(ctx context.Context, claimed DeviceEvalClaimed, deferFor time.Duration) (int64, error) {
|
||
args := []any{claimed.Token, claimed.DeviceID, claimed.Version, time.Now().UnixMilli(), deferFor.Milliseconds()}
|
||
for _, scene := range claimed.Scenes {
|
||
args = append(args, scene)
|
||
}
|
||
return deviceFinishScript.Run(ctx, s.redis,
|
||
[]string{constants.RedisDeviceEvalClaimKey(claimed.DeviceID), constants.RedisDeviceEvalLockKey(claimed.DeviceID),
|
||
constants.RedisDeviceEvalRunningKey, constants.RedisDeviceEvalVersionKey(claimed.DeviceID),
|
||
constants.RedisDeviceEvalScenesKey(claimed.DeviceID), constants.RedisDeviceEvalQueueKey},
|
||
args...).Int64()
|
||
}
|
||
|
||
// ReleaseDeviceEval 发布失败时释放设备判断处理中标记。
|
||
func (s *Store) ReleaseDeviceEval(ctx context.Context, deviceID uint, token string) error {
|
||
return srReleaseScript.Run(ctx, s.redis,
|
||
[]string{constants.RedisDeviceEvalClaimKey(deviceID), constants.RedisDeviceEvalRunningKey, constants.RedisDeviceEvalQueueKey},
|
||
token, deviceID, time.Now().UnixMilli()).Err()
|
||
}
|
||
|
||
// StopResumeQueueStats 是账户停复机队列的即时统计。
|
||
type StopResumeQueueStats struct {
|
||
Queued int64
|
||
Due int64
|
||
Running int64
|
||
Pending int64
|
||
}
|
||
|
||
// StopResumeStats 读取账户停复机队列长度、到期数与执行中数；待确认数按队列成员抽样统计（最多 500 个）。
|
||
func (s *Store) StopResumeStats(ctx context.Context, carrierID uint) (StopResumeQueueStats, error) {
|
||
now := strconv.FormatInt(time.Now().UnixMilli(), 10)
|
||
queueKey := constants.RedisCarrierStopResumeQueueKey(carrierID)
|
||
pipe := s.redis.Pipeline()
|
||
queued := pipe.ZCard(ctx, queueKey)
|
||
due := pipe.ZCount(ctx, queueKey, "-inf", now)
|
||
running := pipe.ZCount(ctx, constants.RedisCarrierStopResumeRunningKey(carrierID), now, "+inf")
|
||
members := pipe.ZRange(ctx, queueKey, 0, 499)
|
||
if _, err := pipe.Exec(ctx); err != nil && !stderrors.Is(err, redis.Nil) {
|
||
return StopResumeQueueStats{}, err
|
||
}
|
||
stats := StopResumeQueueStats{Queued: queued.Val(), Due: due.Val(), Running: running.Val()}
|
||
if ids := members.Val(); len(ids) > 0 {
|
||
check := s.redis.Pipeline()
|
||
results := make([]*redis.StringCmd, 0, len(ids))
|
||
for _, id := range ids {
|
||
cardID, convErr := strconv.ParseUint(strings.TrimSpace(id), 10, 64)
|
||
if convErr != nil {
|
||
continue
|
||
}
|
||
results = append(results, check.HGet(ctx, constants.RedisCardStopResumeCmdKey(uint(cardID)), srFieldPending))
|
||
}
|
||
_, _ = check.Exec(ctx)
|
||
for _, result := range results {
|
||
if result.Val() == "1" {
|
||
stats.Pending++
|
||
}
|
||
}
|
||
}
|
||
return stats, nil
|
||
}
|
||
|
||
// IncrStopResumeStat 累加账户停复机执行计数（10 分钟分桶，保留 2 小时）。
|
||
func (s *Store) IncrStopResumeStat(ctx context.Context, carrierID uint, field string) {
|
||
key := constants.RedisCarrierStopResumeStatKey(carrierID, time.Now().Unix()/constants.StopResumeStatBucketSeconds)
|
||
pipe := s.redis.Pipeline()
|
||
pipe.HIncrBy(ctx, key, field, 1)
|
||
pipe.Expire(ctx, key, 2*time.Hour)
|
||
_, _ = pipe.Exec(ctx)
|
||
}
|
||
|
||
// StopResumeStatLastHour 汇总账户近 1 小时（最近 6 个分桶）的停复机执行计数。
|
||
func (s *Store) StopResumeStatLastHour(ctx context.Context, carrierID uint) map[string]int64 {
|
||
current := time.Now().Unix() / constants.StopResumeStatBucketSeconds
|
||
pipe := s.redis.Pipeline()
|
||
cmds := make([]*redis.MapStringStringCmd, 0, 6)
|
||
for bucket := current - 5; bucket <= current; bucket++ {
|
||
cmds = append(cmds, pipe.HGetAll(ctx, constants.RedisCarrierStopResumeStatKey(carrierID, bucket)))
|
||
}
|
||
_, _ = pipe.Exec(ctx)
|
||
totals := map[string]int64{}
|
||
for _, cmd := range cmds {
|
||
for field, value := range cmd.Val() {
|
||
count, _ := strconv.ParseInt(value, 10, 64)
|
||
totals[field] += count
|
||
}
|
||
}
|
||
return totals
|
||
}
|