基于 Redis 实现分布式锁 🔒 - Golang

type Client struct {
	client redis.Cmdable      // Redis 客户端接口
	g      singleflight.Group // 用于防止缓存击穿的工具，可以让并发请求变成单个请求
	valuer func() string      // 用于生成唯一的值，确保锁的持有者唯一（通常是 UUID）
}
type Lock struct {
	client           redis.Cmdable // 持有的 Redis 客户端，用于操作锁
	key              string        // 锁的键，用于标识锁
	value            string        // 锁的值，用于标识持有者，避免误释放其他客户端的锁
	expiration       time.Duration // 锁的过期时间
	unlock           chan struct{} // 用于信号化锁的释放操作，退出自动续约逻辑
	signalUnlockOnce sync.Once     // 确保解锁操作只执行一次，防止重复释放
}

type RetryStrategy interface {
	// Next 返回下一次重试的间隔，如果不需要继续重试，那么第二参数发挥 false
	Next() (time.Duration, bool)
}

type FixIntervalRetry struct {
	// 重试间隔
	Interval time.Duration
	// 最大次数
	Max int
	cnt int
}

func (f *FixIntervalRetry) Next() (time.Duration, bool) {
	f.cnt++
	return f.Interval, f.cnt <= f.Max
}

func (c *Client) Lock(ctx context.Context, key string, expiration time.Duration, retry RetryStrategy, timeout time.Duration) (*Lock, error) {
    val := c.valuer()  // 生成唯一的锁值，标识当前客户端
    var timer *time.Timer
    defer func() {
        if timer != nil {
            timer.Stop()  // 在退出之前停止定时器，避免资源泄露
        }
    }()
    for {
        lctx, cancel := context.WithTimeout(ctx, timeout)  // 为每次获取锁尝试设置超时
        res, err := c.client.Eval(lctx, luaLock, []string{key}, val, expiration.Seconds()).Result()
        cancel()
        if err != nil && !errors.Is(err, context.DeadlineExceeded) {
            // 非超时错误，比如 Redis server 崩溃、网络异常等，不再继续尝试
            return nil, err
        }
        if res == "OK" {
            // 锁获取成功，返回锁对象
            return newLock(c.client, key, val, expiration), nil
        }
        interval, ok := retry.Next()
        if !ok {
            // 如果重试次数耗尽，返回错误
            if err != nil {
                err = fmt.Errorf("最后一次重试错误: %w", err)
            } else {
                err = fmt.Errorf("锁被人持有: %w", ErrFailedToPreemptLock)
            }
            return nil, fmt.Errorf("rlock: 重试机会耗尽，%w", err)
        }
        // 设置重试的等待时间
        if timer == nil {
            timer = time.NewTimer(interval)
        } else {
            timer.Reset(interval)
        }
        // 等待重试或者上下文结束
        select {
        case <-timer.C: // 定时器到达，表示重试间隔已过，可以再次尝试获取锁
        case <-ctx.Done(): // 上下文 ctx 被取消
            return nil, ctx.Err()
        }
    }
}

func (l *Lock) Unlock(ctx context.Context) error {
	// 使用 Lua 脚本进行解锁操作，保证原子性
	res, err := l.client.Eval(ctx, luaUnlock, []string{l.key}, l.value).Int64()

	defer func() {
		// 确保只进行一次解锁信号操作，防止重复解锁引起 panic
		l.signalUnlockOnce.Do(func() {
			l.unlock <- struct{}{}  // 向 unlock 通道发送信号
			close(l.unlock)         // 关闭 unlock 通道
		})
	}()

	if err == redis.Nil {
		// 锁不存在或者锁的值与预期不符，表示当前锁不属于调用者
		return ErrLockNotHold
	}
	if err != nil {
		// Redis 操作出错，返回错误
		return err
	}
	if res != 1 {
		// Lua 脚本执行结果不是 1，表示锁未被成功释放
		return ErrLockNotHold
	}
	return nil  // 成功释放锁
}

func (l *Lock) Refresh(ctx context.Context) error {
    // 使用 Lua 脚本执行锁续约操作，保证续约的原子性
    res, err := l.client.Eval(ctx, luaRefresh,
        []string{l.key}, l.value, l.expiration.Seconds()).Int64()
    
    if err != nil {
        // Redis 操作出错，返回错误
        return err
    }
    if res != 1 {
        // Lua 脚本执行结果不是 1，表示锁续约失败，可能是锁不存在或者锁的值不匹配
        return ErrLockNotHold
    }
    return nil  // 成功续约锁
}

func (l *Lock) AutoRefresh(interval time.Duration, timeout time.Duration) error {
    // 创建一个定时器，每隔 interval 时间触发一次
    ticker := time.NewTicker(interval)
    // 刷新超时 channel，用于通知在发生超时时重试
    ch := make(chan struct{}, 1)
    defer func() {
        // 确保在函数退出时停止定时器并关闭通道
        ticker.Stop()
        close(ch)
    }()

    for {
        select {
        case <-ticker.C:
            // 定时器触发时创建一个带有超时时间的上下文，用于刷新锁
            ctx, cancel := context.WithTimeout(context.Background(), timeout)
            err := l.Refresh(ctx)
            cancel()

            // 如果刷新操作超时，则继续尝试
            if err == context.DeadlineExceeded {
                // 因为有两个地方可能要写入数据到 ch，而 ch 容量只有 1
                // 如果写入不成功，则说明前一次调用超时且尚未被处理
                select {
                case ch <- struct{}{}:
                default:
                }
                continue
            }

            // 如果刷新过程中出现了其他错误，则返回错误
            if err != nil {
                return err
            }

        case <-ch:
            // 超时处理，重新尝试刷新锁
            ctx, cancel := context.WithTimeout(context.Background(), timeout)
            err := l.Refresh(ctx)
            cancel()

            // 如果仍然超时，则继续重试
            if err == context.DeadlineExceeded {
                select {
                case ch <- struct{}{}:
                default:
                }
                continue
            }

            // 处理其他错误
            if err != nil {
                return err
            }

        case <-l.unlock:
            // 当检测到 unlock 信号时，退出续约逻辑，说明锁已经被释放
            return nil
        }
    }
}