update

database/redis.go

@@ -7,6 +7,7 @@ import (
 
 	"github.com/dgraph-io/ristretto"
 	"github.com/jinzhu/copier"
+	"github.com/redis/go-redis/v9"
 	"gitlab.wanzhuangkj.com/tush/xpkg/config"
 	"gitlab.wanzhuangkj.com/tush/xpkg/pkg/goredis"
 	"gitlab.wanzhuangkj.com/tush/xpkg/pkg/logger"
@@ -20,6 +21,7 @@ var (
 
 var (
 	redisCli     *goredis.Client
+	redCli       *redis.Client
 	redisCliOnce sync.Once
 	memoryCache  *ristretto.Cache
 	cacheType    string
@@ -82,6 +84,7 @@ func InitRedis() {
 	if err != nil {
 		panic("init redis error: " + err.Error())
 	}
+	redCli = redisCli
 }
 
 func GetRedisCli() *goredis.Client {
@@ -93,6 +96,15 @@ func GetRedisCli() *goredis.Client {
 	return redisCli
 }
 
+func GetRedCli() *redis.Client {
+	if redCli == nil {
+		redisCliOnce.Do(func() {
+			InitRedis()
+		})
+	}
+	return redCli
+}
+
 func CloseRedis() error {
 	return goredis.Close(redisCli)
 }

pkg/goredis/goredis.go

@@ -43,9 +43,9 @@ func Init(dsn string, opts ...Option) (*redis.Client, error) {
 
 	rdb := redis.NewClient(opt)
 
-	if o.logger != nil {
-		rdb.AddHook(NewLogHook(o.logger))
-	}
+	// if o.logger != nil {
+	// 	rdb.AddHook(NewLogHook(o.logger))
+	// }
 
 	if o.tracerProvider != nil {
 		err = redisotel.InstrumentTracing(rdb, redisotel.WithTracerProvider(o.tracerProvider))

pkg/redlock/redlock.go

+package redlock
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	"github.com/go-redsync/redsync/v4"
+	"github.com/go-redsync/redsync/v4/redis/goredis/v9"
+	"github.com/redis/go-redis/v9"
+	"go.uber.org/zap"
+)
+
+var (
+	defaultRedSync *redsync.Redsync
+	logger         *zap.Logger
+)
+
+// Config 锁配置
+type Config struct {
+	RedisAddr     string        `yaml:"redis_addr"`
+	RedisPassword string        `yaml:"redis_password"`
+	RedisDB       int           `yaml:"redis_db"`
+	PoolSize      int           `yaml:"pool_size"`
+	MinIdleConns  int           `yaml:"min_idle_conns"`
+	DialTimeout   time.Duration `yaml:"dial_timeout"`
+}
+
+// Init 初始化分布式锁
+func Init(cfg Config, log *zap.Logger) error {
+	logger = log
+
+	client := redis.NewClient(&redis.Options{
+		Addr:         cfg.RedisAddr,
+		Password:     cfg.RedisPassword,
+		DB:           cfg.RedisDB,
+		PoolSize:     cfg.PoolSize,
+		MinIdleConns: cfg.MinIdleConns,
+		DialTimeout:  cfg.DialTimeout,
+	})
+
+	// 测试连接
+	if err := client.Ping(context.Background()).Err(); err != nil {
+		return fmt.Errorf("redis连接失败: %w", err)
+	}
+
+	pool := goredis.NewPool(client)
+	defaultRedSync = redsync.New(pool)
+
+	logger.Info("分布式锁初始化完成")
+	return nil
+}
+
+// Options 锁选项
+type Options struct {
+	Expiry      time.Duration // 锁过期时间
+	Tries       int           // 尝试次数
+	RetryDelay  time.Duration // 重试延迟
+	Timeout     time.Duration // 获取锁超时时间
+	DriftFactor float64       // 时钟漂移因子
+}
+
+// DefaultOptions 默认选项
+func DefaultOptions() Options {
+	return Options{
+		Expiry:      10 * time.Second,
+		Tries:       3,
+		RetryDelay:  500 * time.Millisecond,
+		Timeout:     5 * time.Second,
+		DriftFactor: 0.01,
+	}
+}
+
+// Mutex 互斥锁
+type Mutex struct {
+	name    string
+	options Options
+}
+
+// NewMutex 创建新的互斥锁
+func NewMutex(name string, opts ...func(*Options)) *Mutex {
+	options := DefaultOptions()
+	for _, opt := range opts {
+		opt(&options)
+	}
+
+	return &Mutex{
+		name:    name,
+		options: options,
+	}
+}
+
+// WithExpiry 设置过期时间
+func WithExpiry(d time.Duration) func(*Options) {
+	return func(o *Options) { o.Expiry = d }
+}
+
+// WithTimeout 设置超时时间
+func WithTimeout(d time.Duration) func(*Options) {
+	return func(o *Options) { o.Timeout = d }
+}
+
+// WithTries 设置尝试次数
+func WithTries(tries int) func(*Options) {
+	return func(o *Options) { o.Tries = tries }
+}
+
+// Lock 获取锁
+func (m *Mutex) Lock(ctx context.Context) (func(), error) {
+	if defaultRedSync == nil {
+		return nil, fmt.Errorf("分布式锁未初始化")
+	}
+
+	// 创建redsync互斥锁
+	rsMutex := defaultRedSync.NewMutex(
+		m.name,
+		redsync.WithExpiry(m.options.Expiry),
+		redsync.WithTries(m.options.Tries),
+		redsync.WithRetryDelay(m.options.RetryDelay),
+		redsync.WithDriftFactor(m.options.DriftFactor),
+	)
+
+	// 设置获取锁的超时上下文
+	lockCtx := ctx
+	var cancel context.CancelFunc
+	if m.options.Timeout > 0 {
+		lockCtx, cancel = context.WithTimeout(ctx, m.options.Timeout)
+	} else {
+		lockCtx, cancel = context.WithCancel(ctx)
+	}
+	defer cancel()
+
+	// 尝试获取锁
+	start := time.Now()
+	if err := rsMutex.LockContext(lockCtx); err != nil {
+		logger.Debug("获取分布式锁失败",
+			zap.String("key", m.name),
+			zap.Error(err),
+			zap.Duration("cost", time.Since(start)))
+		return nil, fmt.Errorf("获取锁失败: %w", err)
+	}
+
+	logger.Debug("获取分布式锁成功",
+		zap.String("key", m.name),
+		zap.Duration("cost", time.Since(start)))
+
+	// 返回释放锁的函数
+	unlock := func() {
+		unlockCtx, cancel := context.WithTimeout(context.Background(), 2*time.Minute)
+		defer cancel()
+
+		if ok, err := rsMutex.UnlockContext(unlockCtx); !ok || err != nil {
+			logger.Warn("释放分布式锁失败",
+				zap.String("key", m.name),
+				zap.Bool("ok", ok),
+				zap.Error(err))
+		} else {
+			logger.Debug("释放分布式锁成功",
+				zap.String("key", m.name))
+		}
+	}
+
+	return unlock, nil
+}
+
+// Do 执行受保护的操作（推荐使用）
+func (m *Mutex) Do(ctx context.Context, fn func() error) error {
+	unlock, err := m.Lock(ctx)
+	if err != nil {
+		return err
+	}
+	defer unlock()
+
+	return fn()
+}

pkg/redlock/redlock_test.go

+package redlock_test
+
+import (
+	"context"
+	"fmt"
+	"sync"
+	"testing"
+	"time"
+
+	"gitlab.wanzhuangkj.com/tush/xpkg/pkg/redlock"
+	"go.uber.org/zap"
+)
+
+func TestLock(t *testing.T) {
+	// 1. 正确初始化日志，处理错误
+	logger, err := zap.NewProduction()
+	if err != nil {
+		panic(fmt.Sprintf("初始化日志失败: %v", err))
+	}
+	defer logger.Sync()
+
+	// 2. 初始化分布式锁
+	//   dsn: "default:o$ka25ENHF6!OiZ8@r-bp1010ih815vsvrksc.redis.rds.aliyuncs.com:6379/1"
+	lockCfg := redlock.Config{
+		// RedisAddr:     "localhost:30079",
+		// RedisPassword: "password123",
+		RedisAddr:     "r-bp1010ih815vsvrksc.redis.rds.aliyuncs.com:6379",
+		RedisPassword: "o$ka25ENHF6!OiZ8",
+		RedisDB:       0,
+		PoolSize:      10,
+		MinIdleConns:  5,
+		DialTimeout:   5 * time.Second,
+	}
+	if err := redlock.Init(lockCfg, logger); err != nil {
+		logger.Fatal("初始化分布式锁失败", zap.Error(err))
+	}
+
+	// 3. 创建一把锁
+	resourceKey := "r:feedback:abc"
+	mutex := redlock.NewMutex(resourceKey,
+		redlock.WithExpiry(10*time.Second), // 锁10秒后自动过期
+		redlock.WithTimeout(3*time.Second), // 关键：设置获取锁的超时时间，避免无限等待
+	)
+
+	var wg sync.WaitGroup
+
+	// 4. 启动第一个goroutine
+	wg.Add(1)
+	go func(id int) {
+		defer wg.Done()
+		fmt.Printf("Goroutine %d: 尝试获取锁...\n", id)
+
+		// 为本次锁操作创建一个带超时的上下文
+		ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+		defer cancel()
+
+		err := mutex.Do(ctx, func() error {
+			fmt.Printf("Goroutine %d: 成功获取锁，开始执行任务\n", id)
+			time.Sleep(2 * time.Second) // 模拟耗时任务
+			fmt.Printf("Goroutine %d: 任务执行完毕\n", id)
+			return nil
+		})
+
+		if err != nil {
+			fmt.Printf("Goroutine %d: 执行失败 - %v\n", id, err)
+		} else {
+			fmt.Printf("Goroutine %d: 执行成功\n", id)
+		}
+	}(1)
+
+	// 5. 等待一小段时间，确保第一个goroutine先拿到锁
+	time.Sleep(100 * time.Millisecond)
+
+	// 6. 启动第二个goroutine（它将等待或超时）
+	wg.Add(1)
+	go func(id int) {
+		defer wg.Done()
+		fmt.Printf("Goroutine %d: 尝试获取锁...\n", id)
+
+		ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+		defer cancel()
+
+		err := mutex.Do(ctx, func() error {
+			fmt.Printf("Goroutine %d: 成功获取锁，开始执行任务\n", id)
+			time.Sleep(1 * time.Second)
+			fmt.Printf("Goroutine %d: 任务执行完毕\n", id)
+			return nil
+		})
+
+		if err != nil {
+			// 这里很可能会失败，因为锁被第一个goroutine持有
+			fmt.Printf("Goroutine %d: 执行失败 - %v\n", id, err)
+		} else {
+			fmt.Printf("Goroutine %d: 执行成功\n", id)
+		}
+	}(2)
+
+	// 7. 等待所有goroutine完成
+	wg.Wait()
+	fmt.Println("测试结束")
+}