分布式锁实现

题目描述

如何使用 Redis 实现分布式锁？需要注意哪些问题？

示例场景

# 场景：秒杀商品
# 多个服务同时扣减库存

# 服务 1
GET product:1:stock  # 100
# 判断库存 > 0
# 扣减库存
SET product:1:stock 99

# 服务 2（同时执行）
GET product:1:stock  # 100
# 判断库存 > 0
# 扣减库存
SET product:1:stock 99

# 问题：超卖！库存应该是 98，实际是 99

提示

• 使用 SET NX 实现加锁
• 设置过期时间防止死锁
• 使用 Lua 脚本保证原子性
• 考虑主从切换、锁续期等问题

解法

# 使用 SET NX EX 命令
SET lock:product:1 "uuid" NX EX 30

# 参数说明
# lock:product:1: 锁的 key
# uuid: 唯一标识（防止误删其他线程的锁）
# NX: 不存在时才设置
# EX 30: 30 秒后自动过期

# 返回值
# OK: 加锁成功
# (nil): 加锁失败

# 使用 Lua 脚本保证原子性
if redis.call("get", KEYS[1]) == ARGV[1] then
    return redis.call("del", KEYS[1])
else
    return 0
end

# 执行
EVAL script 1 lock:product:1 uuid

# 参数说明
# script: Lua 脚本
# 1: KEYS 的数量
# lock:product:1: 锁的 key
# uuid: 唯一标识

package redis

import (
    "context"
    "errors"
    "time"
    "github.com/go-redis/redis/v8"
)

var (
    ErrLockNotHeld = errors.New("lock not held")
)

// DistributedLock 分布式锁
type DistributedLock struct {
    client *redis.Client
    key    string
    value  string // 唯一标识
    ttl    time.Duration
}

// NewDistributedLock 创建分布式锁
func NewDistributedLock(client *redis.Client, key string, value string, ttl time.Duration) *DistributedLock {
    return &DistributedLock{
        client: client,
        key:    key,
        value:  value,
        ttl:    ttl,
    }
}

// Lock 加锁
func (l *DistributedLock) Lock(ctx context.Context) error {
    // SET key value NX EX ttl
    return l.client.SetNX(ctx, l.key, l.value, l.ttl).Err()
}

// Unlock 解锁
func (l *DistributedLock) Unlock(ctx context.Context) error {
    // Lua 脚本：保证原子性
    script := `
        if redis.call("get", KEYS[1]) == ARGV[1] then
            return redis.call("del", KEYS[1])
        else
            return 0
        end
    `
    
    result, err := l.client.Eval(ctx, script, []string{l.key}, l.value).Result()
    if err != nil {
        return err
    }
    
    if result == int64(0) {
        return ErrLockNotHeld
    }
    
    return nil
}

// TryLock 尝试加锁（带重试）
func (l *DistributedLock) TryLock(ctx context.Context, retry int, interval time.Duration) error {
    var err error
    for i := 0; i < retry; i++ {
        err = l.Lock(ctx)
        if err == nil {
            return nil
        }
        
        if i < retry-1 {
            time.Sleep(interval)
        }
    }
    return err
}

// 示例使用
func main() {
    client := redis.NewClient(&redis.Options{
        Addr: "localhost:6379",
    })
    
    lock := NewDistributedLock(
        client,
        "lock:product:1",
        uuid.New().String(),
        30*time.Second,
    )
    
    ctx := context.Background()
    
    // 加锁
    err := lock.TryLock(ctx, 3, 100*time.Millisecond)
    if err != nil {
        fmt.Println("加锁失败:", err)
        return
    }
    defer lock.Unlock(ctx)
    
    // 执行业务逻辑
    fmt.Println("加锁成功，执行业务逻辑")
}

常见问题

1. 锁超时问题

# 问题：业务执行时间超过锁的 TTL
SET lock:product:1 "uuid" NX EX 30
# 业务执行 35 秒
# 30 秒后，锁自动过期
# 其他线程获取锁
# 35 秒后，第一个线程执行完，解锁（删除了其他线程的锁）

# 解决方案：锁续期（Watchdog）
# 启动一个后台协程，定期续期

2. 主从切换问题

# 问题：主节点加锁，数据未同步到从节点
# 主节点故障，从节点升级为主节点
# 其他线程可以获取锁（数据丢失）

# 解决方案：Redlock
# 在多个独立的 Redis 实例上加锁
# 大多数成功才算成功

3. 原子性问题

# 问题：加锁和设置过期时间不是原子操作
SETNX lock:product:1 "uuid"  # 加锁成功
EXPIRE lock:product:1 30      # 设置过期时间失败（进程崩溃）
# 锁永久有效，死锁

# 解决方案：使用 SET NX EX
SET lock:product:1 "uuid" NX EX 30  # 原子操作

4. 误删锁问题

# 问题：线程 A 加锁，业务执行时间过长
# 锁过期，线程 B 加锁
# 线程 A 执行完，解锁（删除了线程 B 的锁）

# 解决方案：唯一标识
# 每个线程设置唯一的 value
# 解锁时检查 value 是否匹配

Redlock 算法

原理

# 在 N 个独立的 Redis 实例上加锁
# 大多数（N/2 + 1）成功才算成功

# 步骤
1. 获取当前时间戳
2. 按顺序在 N 个实例上加锁
3. 计算加锁耗时
4. 如果成功数 >= N/2 + 1 且耗时 < TTL，加锁成功
5. 否则，在所有实例上解锁

# 解锁
# 在所有实例上解锁

代码实现

// Redlock 红锁
type Redlock struct {
    clients []*redis.Client
    quorum  int
    ttl     time.Duration
}

// NewRedlock 创建红锁
func NewRedlock(addrs []string, ttl time.Duration) *Redlock {
    clients := make([]*redis.Client, len(addrs))
    for i, addr := range addrs {
        clients[i] = redis.NewClient(&redis.Options{
            Addr: addr,
        })
    }
    
    return &Redlock{
        clients: clients,
        quorum:  len(addrs)/2 + 1,
        ttl:     ttl,
    }
}

// Lock 加锁
func (r *Redlock) Lock(ctx context.Context, key string, value string) error {
    var success int
    
    for _, client := range r.clients {
        err := client.SetNX(ctx, key, value, r.ttl).Err()
        if err == nil {
            success++
        }
    }
    
    if success < r.quorum {
        r.Unlock(ctx, key, value)
        return errors.New("failed to acquire lock")
    }
    
    return nil
}

// Unlock 解锁
func (r *Redlock) Unlock(ctx context.Context, key string, value string) error {
    var errs []error
    
    for _, client := range r.clients {
        script := `
            if redis.call("get", KEYS[1]) == ARGV[1] then
                return redis.call("del", KEYS[1])
            else
                return 0
            end
        `
        err := client.Eval(ctx, script, []string{key}, value).Err()
        if err != nil {
            errs = append(errs, err)
        }
    }
    
    if len(errs) > 0 {
        return fmt.Errorf("failed to unlock: %v", errs)
    }
    
    return nil
}