Python使用Redis分布式锁时如何设置获取优先级保证FIFO顺序

问题根源 #

原生redis-py库提供的默认锁实现不保障FIFO获取顺序：所有阻塞等待锁的客户端在锁释放时会同时发起抢占请求，率先到达Redis服务端的请求会拿到锁，和客户端提交申请的先后顺序无关，因此会出现先等待的用户后获取锁的乱序问题。

实现方案 #

基于Redis有序集合(ZSET)实现公平FIFO锁，核心逻辑如下：

• 每个锁申请请求生成全局唯一的标识ID，以申请时间戳+随机后缀作为score写入对应锁的ZSET
• 申请者检查自身score是否为ZSET中最小值：如果是则成功获取锁；如果不是则阻塞等待，监听锁释放的广播通知
• 锁释放时，持有者删除自身在ZSET中的记录，同时发布锁释放通知，等待队列中的申请者收到通知后再次检查自身排名，排名第一的申请者获得锁
• 额外增加锁超时自动释放逻辑，避免客户端异常崩溃导致锁永久占用

代码实现示例 #

import redis
import time
import uuid

REDIS_CLIENT = redis.Redis(decode_responses=True)

class FIFORedisLock:
    def __init__(self, lock_key, timeout=60, blocking_timeout=60):
        self.lock_key = lock_key
        self.wait_zset_key = f"{lock_key}:wait_queue"
        self.pubsub_channel = f"{lock_key}:release_notify"
        self.timeout = timeout
        self.blocking_timeout = blocking_timeout
        self.request_id = f"{time.time()}:{uuid.uuid4().hex}"
        self.score = float(self.request_id.split(":")[0])

    def acquire(self):
        start_time = time.time()
        # 写入等待队列
        REDIS_CLIENT.zadd(self.wait_zset_key, {self.request_id: self.score})
        pubsub = REDIS_CLIENT.pubsub()
        pubsub.subscribe(self.pubsub_channel)

        try:
            while time.time() - start_time < self.blocking_timeout:
                # 获取队列中排名第一的请求
                first_request = REDIS_CLIENT.zrange(self.wait_zset_key, 0, 0, withscores=True)
                if first_request and first_request[0][0] == self.request_id:
                    # 成功获取锁，设置过期时间
                    REDIS_CLIENT.expire(self.wait_zset_key, self.timeout)
                    return True
                # 等待释放通知，最多等1秒避免死等
                message = pubsub.get_message(timeout=1)
            # 阻塞超时，清理自身请求记录
            REDIS_CLIENT.zrem(self.wait_zset_key, self.request_id)
            return False
        finally:
            pubsub.close()

    def release(self):
        # 只有锁的持有者可以释放
        first_request = REDIS_CLIENT.zrange(self.wait_zset_key, 0, 0)
        if first_request and first_request[0] == self.request_id:
            REDIS_CLIENT.zrem(self.wait_zset_key, self.request_id)
            # 通知其他等待者
            REDIS_CLIENT.publish(self.pubsub_channel, "release")
            return True
        return False

# 用法和原有锁基本一致
lock = FIFORedisLock("task_key", timeout=60)
have_lock = lock.acquire()
if have_lock:
    try:
        # 运行你的任务逻辑
        pass
    finally:
        lock.release()

Celery场景适配 #

该实现可以直接替换你现有Celery唯一任务逻辑中的原生Redis锁，原有任务调度逻辑不需要做额外修改，即可保障任务执行顺序和用户申请顺序一致。

内容的提问来源于stack exchange，提问作者Rom1