通过lua实现redis 分布式锁的项目实践_Redis

在 redis 中使用 lua 脚本实现分布式锁 是一种高效、原子、安全的方式。由于 lua 脚本在 redis 中是单线程、原子执行的，可以避免多个客户端在加锁/解锁时出现竞态条件（race condition）。

下面将为你提供一个 完整、生产可用的 redis 分布式锁 lua 实现方案，包含：

加锁（lock）
解锁（unlock）
自动续期（可选）
防止误删他人锁
支持可重入（可选）

✅ 一、核心设计原则

锁的唯一标识：每个客户端使用唯一的 value（如 uuid）作为锁标识，防止 a 客户端释放 b 客户端的锁。
原子性操作：加锁和解锁必须通过 lua 脚本保证原子性。
自动过期：设置 ttl（如 30 秒），防止死锁。
可重入支持（可选） ：同一个客户端可多次加锁（需记录重入次数）。

本文先实现 基础版（不可重入） ，再给出 可重入增强版。

✅ 二、基础版：不可重入分布式锁

🔐 1. 加锁脚本（lock.lua）

-- keys[1] = 锁的 key（如 "my_lock"）
-- argv[1] = 锁的 value（唯一标识，如 uuid）
-- argv[2] = 过期时间（毫秒，如 "30000"）

local key = keys[1]
local value = argv[1]
local ttl = tonumber(argv[2])

-- 尝试加锁：只有当 key 不存在时才设置成功
if redis.call("set", key, value, "nx", "px", ttl) then
    return 1  -- 加锁成功
else
    return 0  -- 加锁失败（已被占用）
end

✅ set key value nx px ttl 是 redis 2.6.12+ 的原子命令：

nx：仅当 key 不存在时设置
px：设置过期时间（毫秒）

🔓 2. 解锁脚本（unlock.lua）

-- keys[1] = 锁的 key
-- argv[1] = 锁的 value（用于验证是否是自己的锁）

local key = keys[1]
local value = argv[1]

-- 获取当前锁的 value
local current_value = redis.call("get", key)

-- 如果存在且等于自己的 value，则删除
if current值 == value then
    redis.call("del", key)
    return 1  -- 解锁成功
else
    return 0  -- 不是自己的锁，拒绝删除
end

⚠️ 注意：不能直接 del，否则可能误删别人的锁！

✅ 三、python 示例（使用 redis-py）

import redis
import uuid
import time

# 初始化 redis 连接
r = redis.strictredis(host='localhost', port=6379, decode_responses=true)

# 加载 lua 脚本（推荐使用 script_load + evalsha 提高性能）
lock_script = r.register_script("""
if redis.call("set", keys[1], argv[1], "nx", "px", argv[2]) then
    return 1
else
    return 0
end
""")

unlock_script = r.register_script("""
if redis.call("get", keys[1]) == argv[1] then
    return redis.call("del", keys[1])
else
    return 0
end
""")

def acquire_lock(lock_name, acquire_timeout=10, lock_timeout=30):
    """获取分布式锁"""
    identifier = str(uuid.uuid4())
    lock_timeout_ms = int(lock_timeout * 1000)
    end_time = time.time() + acquire_timeout

    while time.time() < end_time:
        # 执行 lua 脚本尝试加锁
        if lock_script(keys=[lock_name], args=[identifier, lock_timeout_ms]):
            return identifier  # 返回锁标识
        time.sleep(0.01)  # 短暂等待后重试

    return false

def release_lock(lock_name, identifier):
    """释放分布式锁"""
    unlock_script(keys=[lock_name], args=[identifier])

# ===== 使用示例 =====
lock_key = "order:123:lock"

# 尝试获取锁
lock_id = acquire_lock(lock_key, acquire_timeout=5, lock_timeout=30)
if lock_id:
    try:
        print("✅ 获取锁成功，执行业务逻辑...")
        time.sleep(5)  # 模拟业务处理
    finally:
        # 释放锁
        release_lock(lock_key, lock_id)
        print("🔓 锁已释放")
else:
    print("❌ 获取锁超时")

✅ 四、增强版：支持可重入锁（reentrant lock）

适用于同一线程/协程需要多次进入临界区的场景。

设计思路：

使用 hash 结构 存储锁：
- field = 客户端 id（uuid）
- value = 重入次数（counter）
设置整体 ttl

🔐 可重入加锁脚本（reentrant_lock.lua）

-- keys[1] = 锁 key
-- argv[1] = 客户端 id
-- argv[2] = 锁过期时间（毫秒）

local key = keys[1]
local client_id = argv[1]
local ttl = tonumber(argv[2])

-- 检查是否已持有锁
local current_client = redis.call("hget", key, client_id)

if current_client then
    -- 已持有：重入次数 +1，刷新 ttl
    redis.call("hincrby", key, client_id, 1)
    redis.call("pexpire", key, ttl)
    return 1
else
    -- 未持有：尝试获取锁
    if redis.call("hlen", key) == 0 then
        -- 锁空闲，创建
        redis.call("hmset", key, client_id, 1)
        redis.call("pexpire", key, ttl)
        return 1
    else
        -- 锁被他人持有
        return 0
    end
end

🔓 可重入解锁脚本（reentrant_unlock.lua）

-- keys[1] = 锁 key
-- argv[1] = 客户端 id

local key = keys[1]
local client_id = argv[1]

-- 检查是否持有锁
local counter = redis.call("hget", key, client_id)

if not counter then
    return 0  -- 未持有锁
end

counter = tonumber(counter)

if counter > 1 then
    -- 重入次数 >1，减1
    redis.call("hincrby", key, client_id, -1)
    return 1
else
    -- 最后一次，删除整个 key
    redis.call("del", key)
    return 1
end

💡 使用方式与基础版类似，只需替换 lua 脚本。

✅ 五、注意事项与最佳实践

问题	解决方案
锁过期但业务未完成	使用“看门狗”（watchdog）线程自动续期（如 redisson 的机制）
主从切换导致锁丢失	使用 redlock 算法（多 redis 实例），但争议较大；更推荐强一致存储（如 etcd）
lua 脚本调试困难	在开发环境打印日志（redis.log(redis.log_warning, "msg")）
性能优化	使用 script load + evalsha 避免每次传输脚本