feat: 添加分布式锁以防止多个任务同时执行

src/framework/cron/cron.go

@@ -1,6 +1,7 @@
 package cron
 
 import (
+	"errors"
 	"fmt"
 	"time"
 
@@ -194,6 +195,9 @@ func (qj QueueJob) Run() {
 	// 获取队列处理器接口实现
 	processor := *job.queueProcessor
 	result, err := processor.Execute(job.Data)
+	if errors.Is(err, ErrTaskRunning) {
+		return
+	}
 	if err != nil {
 		job.Status = Failed
 		cronLog.Error(err, "failed", job)

src/framework/cron/cron_log.go

@@ -2,6 +2,7 @@ package cron
 
 import (
 	"encoding/json"
+	"errors"
 	"time"
 
 	"be.ems/src/framework/constants"
@@ -123,3 +124,6 @@ type JobData struct {
 	// 定时任务调度表记录信息
 	SysJob monitorModel.SysJob
 }
+
+// ErrTaskRunning 任务正在运行错误
+var ErrTaskRunning = errors.New("task is running")

src/framework/database/redis/expand.go

@@ -4,6 +4,8 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"net"
+	"os"
 	"sync"
 	"time"
 
@@ -173,3 +175,49 @@ func Expire(source, key string, expiration time.Duration) error {
 	}
 	return nil
 }
+
+// SetNX 设置分布式锁，不存在时设置
+func SetNX(source, key string, expiration time.Duration) bool {
+	// 数据源
+	rdb := RDB(source)
+	if rdb == nil {
+		return false
+	}
+	hostname, ipAddr := hostnameAndIP()
+	value := fmt.Sprintf("%s@%s", hostname, ipAddr)
+	// 过期时间设置
+	ctx := context.Background()
+	ok, err := rdb.SetNX(ctx, key, value, expiration).Result()
+	if err != nil {
+		logger.Errorf("redis SetNX err %v", err)
+		return false
+	}
+	return ok
+}
+
+// hostnameAndIP 获取当前主机的名称和 IP 地址
+func hostnameAndIP() (string, string) {
+	hostname, err := os.Hostname()
+	if err != nil {
+		return "unknown", "unknown"
+	}
+
+	// 获取本机的第一个非 loopback 的 IP 地址
+	addrs, err := net.InterfaceAddrs()
+	if err != nil {
+		return hostname, "unknown"
+	}
+	var ipAddr string
+	for _, addr := range addrs {
+		// 只选择 IPv4 地址且不是回环地址
+		ipNet, ok := addr.(*net.IPNet)
+		if ok && ipNet.IP.To4() != nil && !ipNet.IP.IsLoopback() {
+			ipAddr = ipNet.IP.String()
+			break
+		}
+	}
+	if ipAddr == "" {
+		return hostname, "unknown"
+	}
+	return hostname, ipAddr
+}