带有乐观锁的事务

WATCH key [key ...]

redis> WATCH user_id_counter
OK
redis> GET user_id_counter -- 获取当前最新的用户ID
"256"
redis> MULTI
OK
redis> SET user::256::email "peter@spamer.com" -- 尝试使用这个ID来存储用户信息
QUEUED
redis> SET user::256::password "topsecret"
QUEUED
redis> INCR user_id_counter -- 创建新的用户ID
QUEUED
redis> EXEC
(nil) -- user_id_counter键已被修改

UNWATCH

redis> WATCH "lock_key" "user_id_counter" "msg"
OK
redis> UNWATCH -- 取消对以上3个键的监视
OK

from redis import WatchError

class IdentityLock:

    def __init__(self, client, key):
        self.client = client
        self.key = key

    def acquire(self, identity, timeout):
        """
        尝试获取一个带有身份标识符和最大使用时限的锁，
        成功时返回 True ，失败时返回 False 。
        """
        result = self.client.set(self.key, identity, ex=timeout, nx=True)
        return result is not None

    def release(self, input_identity):
        """
        根据给定的标识符，尝试释放锁。
        返回 True 表示释放成功；
        返回 False 则表示给定的标识符与锁持有者的标识符并不相同，释放请求被拒绝。
        """
        # 开启流水线
        pipe = self.client.pipeline()
        try:
            # 监视锁键
            pipe.watch(self.key)
            # 获取锁键储存的标识符
            lock_identity = pipe.get(self.key)
            if lock_identity is None:
                # 如果锁键的标识符为空，那么说明锁已经被释放
                return True
            elif input_identity == lock_identity:
                # 如果给定的标识符与锁键储存的标识符相同，那么释放这个锁
                # 为了确保 DEL 命令在执行时的安全性，我们需要使用事务去包裹它
                pipe.multi()
                pipe.delete(self.key)
                pipe.execute()
                return True
            else:
                # 如果给定的标识符与锁键储存的标识符并不相同
                # 那么说明当前客户端不是锁的持有者
                # 拒绝本次释放请求
                return False
        except WatchError:
            # 抛出异常说明在 DEL 命令执行之前，已经有其他客户端修改了锁键
            return False
        finally:
            # 取消对键的监视
            pipe.unwatch()
            # 因为 redis-py 在执行 WATCH 命令期间，会将流水线与单个连接进行绑定
            # 所以在执行完 WATCH 命令之后，必须调用 reset() 方法将连接归还给连接池
            pipe.reset()

>>> from redis import Redis
>>> from identity_lock import IdentityLock
>>> client = Redis(decode_responses=True)
>>> lock = IdentityLock(client, "test-lock")
>>> lock.acquire("peter", 3600) # 使用"peter"作为标识符，获取一个使用时限为3600s的锁
True
>>> lock.release("tom") # 尝试使用错误的标识符去释放锁，失败
False
>>> lock.release("peter") # 使用正确的标识符去释放锁，成功
True

from redis import WatchError

class Semaphore:

    def __init__(self, client, name):
        self.client = client
        self.name = name
        # 用于储存信号量持有者标识符的集合
        self.holder_key = "semaphore::{0}::holders".format(name)
        # 用于记录信号量最大可获取数量的字符串
        self.size_key = "semaphore::{0}::max_size".format(name)

    def set_max_size(self, size):
        """
        设置信号量的最大可获取数量。
        """
        self.client.set(self.size_key, size)

    def get_max_size(self):
        """
        返回信号量的最大可获取数量。
        """
        result = self.client.get(self.size_key)
        if result is None:
            return 0
        else:
            return int(result)

    def get_current_size(self):
        """
        返回目前已被获取的信号量数量。
        """
        return self.client.scard(self.holder_key)

    def acquire(self, identity):
        """
        尝试获取一个信号量，成功时返回 True ，失败时返回 False 。
        传入的 identity 参数将被用于标识客户端的身份。

        如果调用该方法时信号量的最大可获取数量尚未被设置，那么引发一个 TypeError 。
        """
        # 开启流水线
        pipe = self.client.pipeline()
        try:
            # 监视与信号量有关的两个键
            pipe.watch(self.size_key, self.holder_key)

            # 取得当前已被获取的信号量数量，以及最大可获取的信号量数量
            current_size = pipe.scard(self.holder_key)
            max_size_in_str = pipe.get(self.size_key)
            if max_size_in_str is None:
                raise TypeError("Semaphore max size not set")
            else:
                max_size = int(max_size_in_str)

            if current_size < max_size:
                # 如果还有剩余的信号量可用
                # 那么将给定的标识符放入到持有者集合中
                pipe.multi()
                pipe.sadd(self.holder_key, identity)
                pipe.execute()
                return True
            else:
                # 没有信号量可用，获取失败
                return False
        except WatchError:
            # 获取过程中有其他客户端修改了 size_key 或者 holder_key ，获取失败
            return False
        finally:
            # 取消监视
            pipe.unwatch()
            # 将连接归还给连接池
            pipe.reset()

    def release(self, identity):
        """
        根据给定的标识符，尝试释放当前客户端持有的信号量。
        返回 True 表示释放成功，返回 False 表示由于标识符不匹配而导致释放失败。
        """
        # 尝试从持有者集合中移除给定的标识符
        result = self.client.srem(self.holder_key, identity)
        # 移除成功则说明信号量释放成功
        return result == 1

>>> from redis import Redis
>>> from semaphore import Semaphore
>>> client = Redis(decode_responses=True)
>>> semaphore = Semaphore(client, "test-semaphore") # 创建计数信号量
>>> semaphore.set_max_size(3) # 设置信号量的最大可获取数量
>>> semaphore.acquire("peter") # 获取信号量
True
>>> semaphore.acquire("jack")
True
>>> semaphore.acquire("tom")
True
>>> semaphore.acquire("mary") # 可用的3个信号量都已被获取，无法取得更多信号量
False
>>> semaphore.release("jack") # 释放一个信号量
True
>>> semaphore.get_current_size() # 目前有两个信号量已被获取
2
>>> semaphore.get_max_size() # 信号量的最大可获取数量为3个
3