对Redis锁的简单封装

背景

最近遇到了一个需要使用独占锁来保证业务正确性的场景，鉴于服务本身也会使用到 redis 缓存，可以直接利用 redis 提供的锁支持。

基本使用

创建锁

from redis import Redis

client = Redis()
lock = client.lock(name="key",timeout=60.0,blocking_timeout=5.0)

可以直接调用 redis 实例的 lock 方法，并指定锁的名称，超时时间和等待时间（如果未能在 blocking_timeout 内获取到锁，会抛出异常）。

使用锁

1
2
3

lock.acquire()
do_something()
lock.release()

调用acquire方法获取锁，业务逻辑执行完成后调用release方法释放锁。

使用上下文管理器

手动获取并释放锁的使用方法比较繁琐，并且忘记调用acquire方法或因为业务逻辑异常导致acquire方法没有成功调用的风险。我们可以使用上下文管理器来更方便也更安全地使用 redis 锁。

1 2	with client.lock(name="key",timeout=60.0,blocking_timeout=5.0): do_something()

这本身也是 RAII（资源获取即初始化）思想的一个应用。

封装

我们可以使用functools.partial函数，对 redis 的 lock 使用进行一些简单的封装。

from functools import partial

def lock(
    *,
    prefix: str,
    key: str,
    timeout: float,
    blocking: float = 5.0,
    client: Redis = default_client,
) -> Lock:
    return client.lock(f"{prefix}:{key}", timeout=timeout, blocking_timeout=blocking)


a_lock = partial(lock, prefix="A", timeout=60.0)

def do_a(uid: int) -> None:
    with alock(key=uid):
        pass

我们定义了一个lock函数，可以指定使用锁时的名称（由prefix和key组合而来）、超时时间、等待时间和使用的 redis 实例。
在具体的业务场景中，可以使用functools.partial函数定义更个性化的锁。
在这个示例中，我们定义了一个a_lock函数，指定了锁的名称前缀为A，超时时间为 60 秒。
在使用a_lock的时候，只需要指定key 即可。

总结

functools.partial实用性很高，值得在业务中使用。