diff --git a/backend/cmd/api/main.go b/backend/cmd/api/main.go index 59296f5..d9b1dd5 100644 --- a/backend/cmd/api/main.go +++ b/backend/cmd/api/main.go @@ -83,7 +83,7 @@ func main() { jobCtx, stopJobs := context.WithCancel(context.Background()) defer stopJobs() if deps.DB != nil { - ordertimeout.New(deps.DB, logger).Start(jobCtx) + ordertimeout.New(deps.DB, deps.Redis, logger).Start(jobCtx) } server := &http.Server{ Addr: cfg.AppAddr, diff --git a/backend/internal/jobs/ordertimeout/job.go b/backend/internal/jobs/ordertimeout/job.go index 6f633f0..8c0f7a9 100644 --- a/backend/internal/jobs/ordertimeout/job.go +++ b/backend/internal/jobs/ordertimeout/job.go @@ -2,23 +2,31 @@ package ordertimeout import ( "context" + "crypto/rand" + "encoding/hex" "encoding/json" + "fmt" "strconv" "time" "hfb_sys/backend/internal/model" "hfb_sys/backend/internal/modules/notification" + "github.com/redis/go-redis/v9" "go.uber.org/zap" "gorm.io/datatypes" "gorm.io/gorm" "gorm.io/gorm/clause" ) +const orderTimeoutLockKey = "hfb:job:ordertimeout:lock" + type Job struct { - db *gorm.DB - logger *zap.Logger - interval time.Duration + db *gorm.DB + redis *redis.Client + logger *zap.Logger + interval time.Duration + instanceID string } type thresholds struct { @@ -29,14 +37,58 @@ type thresholds struct { OwnerReturnConfirmTimeoutMinutes int } -func New(db *gorm.DB, logger *zap.Logger) *Job { +func New(db *gorm.DB, redisClient *redis.Client, logger *zap.Logger) *Job { return &Job{ - db: db, - logger: logger, - interval: time.Minute, + db: db, + redis: redisClient, + logger: logger, + interval: time.Minute, + instanceID: newInstanceID(), } } +// acquireLock 通过 Redis 分布式锁确保同一时刻只有一个实例执行超时扫描。 +// 未配置 Redis 时直接执行;Redis 出错时降级执行(事务内行锁与状态二次校验可兜底,不会写坏数据)。 +func (j *Job) acquireLock(ctx context.Context) (func(), bool) { + if j.redis == nil { + return func() {}, true + } + ok, err := j.redis.SetNX(ctx, orderTimeoutLockKey, j.instanceID, j.lockTTL()).Result() + if err != nil { + j.logger.Warn("order timeout job acquire lock failed, run without lock", zap.Error(err)) + return func() {}, true + } + if !ok { + return nil, false + } + return j.releaseLock, true +} + +// releaseLock 仅在锁仍归本实例时释放,避免误删其他实例已续上的锁。 +func (j *Job) releaseLock() { + if j.redis == nil { + return + } + relCtx, cancel := context.WithTimeout(context.Background(), 3*time.Second) + defer cancel() + script := redis.NewScript(`if redis.call("get", KEYS[1]) == ARGV[1] then return redis.call("del", KEYS[1]) else return 0 end`) + if err := script.Run(relCtx, j.redis, []string{orderTimeoutLockKey}, j.instanceID).Err(); err != nil { + j.logger.Warn("order timeout job release lock failed", zap.Error(err)) + } +} + +func (j *Job) lockTTL() time.Duration { + return 2 * j.interval +} + +func newInstanceID() string { + b := make([]byte, 8) + if _, err := rand.Read(b); err != nil { + return fmt.Sprintf("inst-%d", time.Now().UnixNano()) + } + return hex.EncodeToString(b) +} + func (j *Job) Start(ctx context.Context) { if j == nil || j.db == nil { return @@ -60,6 +112,12 @@ func (j *Job) loop(ctx context.Context) { } func (j *Job) run(ctx context.Context) { + release, ok := j.acquireLock(ctx) + if !ok { + return + } + defer release() + cfg, err := j.loadThresholds(ctx) if err != nil { j.logger.Warn("order timeout job config load failed", zap.Error(err))