超时任务多实例优化

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,

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