如何让进程释放非自身持有的Redis锁？读写锁实现求助

问题核心原因 #

redis-py提供的lock.Lock是基于持有者令牌验证实现的：只有成功获取锁的客户端（持有该锁对应的唯一令牌），才能调用release()方法释放锁。你的代码中，第一个读进程会获取写锁，但最后一个读进程尝试释放写锁时，并没有持有该锁的令牌，因此触发错误。

解决方案：用Redis Lua脚本实现原子操作 #

Redis的Lua脚本执行是单线程原子性的，我们可以用脚本把「递减读计数器」和「释放写锁」的逻辑合并，直接在Redis端完成，无需依赖客户端的令牌验证（redis-py的Lock底层通过Redis字符串键存储令牌，删除该键等价于释放锁）。

修改后的完整代码 #

from enum import Enum
from redis import StrictRedis, lock

class Prefix(Enum):
    READ = 'read_lock_'
    WRITE = 'write_lock_'
    READ_COUNTER = 'read_counter_'

class RedisLockParams(Enum):
    TIMEOUT = 60  # 锁的最大存活时间（秒）
    SLEEP_TIME = 0.1  # 获取锁失败后的重试间隔（秒）
    BLOCKING = True  # 是否阻塞等待获取锁
    BLOCKING_TIMEOUT = None  # 最大阻塞等待时间（秒）


class ReadWriteLock:
    def __init__(self, lock_name: str, redis_host: str, redis_port: int, redis_key: str):
        self.__read_lock_name = Prefix.READ.value + lock_name
        self.__write_lock_name = Prefix.WRITE.value + lock_name
        self.__read_counter_key = Prefix.READ_COUNTER.value + lock_name
        self.__cache = StrictRedis(host=redis_host,
                                  port=redis_port,
                                  db=0,
                                  password=redis_key,
                                  ssl=True)
        print(f'ping return:{self.__cache.ping()}')
        # 初始化读计数器（不存在则设为0）
        self.__cache.setnx(self.__read_counter_key, 0)

        # 初始化读锁
        self.__read_lock = lock.Lock(self.__cache,
                                    self.__read_lock_name,
                                    RedisLockParams.TIMEOUT.value,
                                    RedisLockParams.SLEEP_TIME.value,
                                    RedisLockParams.BLOCKING.value,
                                    RedisLockParams.BLOCKING_TIMEOUT.value)
        
        # 初始化写锁
        self.__write_lock = lock.Lock(self.__cache,
                                    self.__write_lock_name,
                                    RedisLockParams.TIMEOUT.value,
                                    RedisLockParams.SLEEP_TIME.value,
                                    RedisLockParams.BLOCKING.value,
                                    RedisLockParams.BLOCKING_TIMEOUT.value)
        
        # 定义释放读锁的Lua脚本
        self.__release_read_script = self.__cache.register_script("""
        -- 获取读锁
        local acquired = redis.call('SET', KEYS[1], ARGV[1], 'NX', 'PX', ARGV[2])
        if not acquired then
            return 0
        end
        -- 递减读计数器
        local counter = redis.call('DECR', KEYS[2])
        -- 若计数器为0，删除写锁键（释放写锁）
        if counter == 0 then
            redis.call('DEL', KEYS[3])
        end
        -- 释放读锁
        redis.call('DEL', KEYS[1])
        return 1
        """)

    def acquire_read_lock(self) -> bool:
        write_lock_acquired = False

        self.__read_lock.acquire()
        read_counter = self.__cache.incr(self.__read_counter_key)
        if read_counter == 1:
            write_lock_acquired = self.__write_lock.acquire()  # 第一个读进程获取写锁，阻塞写操作
        self.__read_lock.release()

        return write_lock_acquired

    def release_read_lock(self):
        # 调用Lua脚本完成原子操作
        self.__release_read_script(
            keys=[self.__read_lock_name, self.__read_counter_key, self.__write_lock_name],
            args=[self.__read_lock.token, int(RedisLockParams.TIMEOUT.value * 1000)]
        )

    def acquire_write_lock(self) -> bool:
        return self.__write_lock.acquire()

    def release_write_lock(self):
        self.__write_lock.release()

关键说明 #

1. Lua脚本原子性：脚本中先获取读锁，确保计数器操作不会被其他进程干扰，递减后判断计数器是否为0，若是则直接删除写锁的Redis键（等同于释放写锁），最后释放读锁，整个过程原子完成。
2. 令牌传递：脚本中使用当前读锁的令牌来获取读锁，保证只有合法的读锁持有者才能执行后续操作。
3. 避免跨进程令牌问题：不再由客户端尝试释放不属于自己的写锁，而是由Redis端直接处理写锁释放逻辑，从根源解决持有者验证问题。

内容的提问来源于stack exchange，提问作者Tuan Le PN