分布式锁:SET NX EX + 看门狗

一句话总结

分布式锁就是跨机器抢一把锁——抢到的人干活,没抢到的等着。Redis 用 SET key value NX EX timeout 一条原子命令搞定加锁,Python 用 redis-pyaioredis 实现。看门狗是后台定时续期——你业务没跑完,锁就不会过期。


0. 这东西跟我有什么关系?

这是你作为 Python AI Agent 后端开发者 几乎一定会遇到的需求:

 
场景 1:分布式定时任务
 
  三个 FastAPI 实例,每天凌晨执行一次"归档旧 Agent 对话记录"。
 
  不加锁 → 三个实例同时跑 → 归档 3 次 → 数据重复甚至错乱 ❌
 
场景 2:RAG 文档处理
 
  用户上传一个 500 页的 PDF,交给一个 Agent 切片处理。
 
  如果用户手滑点了两次上传 → 两个 Agent 同时处理同一个文档 → 冲突 ❌
 
  → 分布式锁保证:同一个 document_id 同一时间只处理一次
 
场景 3:API 限流 / 资源抢占
 
  一个 Agent 调度器管理 10 个 LLM API Key,
 
  任务来了需要抢一个 key 用 → 抢到锁的用 key,没抢到的排队。
 

在你的项目里,分布式锁 ≈ 避免多实例打架。


🌰 理解场景

你开了个奶茶店,店里有唯一一台制冰机。

 
单店(单机):
 
  你一个人做奶茶 → 不需要抢,就你一个人用
 
  就算 3 个人也方便 → "我先用" "好好好你先"
 
连锁店(分布式):
 
  三家分店,共享一台制冰机:
 
    广州分店:"我要用制冰机"
 
    深圳分店:"我也要用"
 
    东莞分店:"我也要"
 
  怎么保证同一时间只有一家在用?→ 分布式锁
 
例子都很土,道理就是这个道理。
 

为什么 Python 的 threading.Lock 不行?

 
threading.Lock → 只锁同一个进程里的线程
 
你的 FastAPI 起了 3 个 docker 容器(3 个进程):
 
  容器 1 的 threading.Lock → 锁容器 1 自己的线程
 
  容器 2 的 threading.Lock → 锁容器 2 自己的线程
 
  两个容器同时干活 → 谁也锁不住谁 ❌
 
分布式锁 → 锁存到 Redis → 所有容器都问 Redis → 全局一把锁 ✅
 

一、SET NX EX —— 最基础的分布式锁

一条命令搞定的原因

 
SET lock:doc:1001 "worker_A" NX EX 30
 
参数大白话作用
NX”这个 key 不存在我才设”互斥——锁被抢了就不能再设
EX 30”30 秒后自动消失”防死锁——持锁者挂了,锁自己会没
value = "worker_A"”写上谁拿的锁”身份标识——解锁时验明正身

为什么必须一条命令?(必手写常问)

 
❌ 错误做法 —— 两条命令:
 
  SETNX lock:doc:1001 "worker_A"   -- 第 1 条:加锁
 
  EXPIRE lock:doc:1001 30          -- 第 2 条:设过期
 
  → 第 1 条执行完,Redis 宕机了,EXPIRE 没设上 → 锁永远不释放 → 死锁 ❌
 
✅ 正确做法 —— 一条命令:
 
  SET lock:doc:1001 "worker_A" NX EX 30
 
  → Redis 保证 NX + EX 要么全成,要么全不成 → 不会死锁 ✅
 
道理:两条命令之间可能挂,一条命令不存在"之间"。
 

二、Python 手写实现(必手写)

同步版(redis-py)

 
import redis
 
import uuid
 
r = redis.Redis(host='localhost', port=6379)
 
def lock_and_work(order_id):
 
    lock_key = f"lock:order:{order_id}"
 
    lock_value = str(uuid.uuid4())  # 唯一标识,防误删
 
    # ===== 加锁 =====
 
    locked = r.set(lock_key, lock_value, nx=True, ex=30)
 
    # nx=True 对应 NX,ex=30 对应 EX 30
 
    # 返回 True = 抢到锁,False = 没抢到
 
    if not locked:
 
        raise Exception("系统繁忙,请稍后重试")
 
    try:
 
        # 执行业务
 
        process_order(order_id)
 
    finally:
 
        # ===== 解锁(必须 Lua 脚本)=====
 
        # 先检查 value 是不是自己的,是才删
 
        # 不这样搞的话:你的锁到期了,别人抢了锁,你一把 DEL 把别人的删了
 
        lua_script = """
 
        if redis.call('get', KEYS[1]) == ARGV[1] then
 
            return redis.call('del', KEYS[1])
 
        else
 
            return 0
 
        end
 
        """
 
        r.eval(lua_script, 1, lock_key, lock_value)
 

解锁为什么不用 Python 的 if + delete?

 
Python 写法:
 
  if r.get(lock_key) == lock_value:
 
      r.delete(lock_key)
 
  → GET + DEL 是两条命令
 
  → 刚 GET 完、还没 DEL,锁到期了,别人抢了
 
  → 你继续 DEL → 删了别人的锁 ❌
 
Lua 脚本 → 在 Redis 服务端原子执行 → 不存在"中间被人插一脚" ✅
 

异步版(aioredis / redis-py async)

你的 FastAPI 项目用异步,所以这个要会:

 
import uuid
 
import asyncio
 
from redis.asyncio import Redis
 
async def lock_and_work(order_id: str):
 
    r = Redis(host='localhost', port=6379, decode_responses=True)
 
    lock_key = f"lock:order:{order_id}"
 
    lock_value = str(uuid.uuid4())
 
    locked = await r.set(lock_key, lock_value, nx=True, ex=30)
 
    if not locked:
 
        raise Exception("系统繁忙,请稍后重试")
 
    try:
 
        await process_order(order_id)
 
    finally:
 
        lua_script = """
 
        if redis.call('get', KEYS[1]) == ARGV[1] then
 
            return redis.call('del', KEYS[1])
 
        else
 
            return 0
 
        end
 
        """
 
        await r.eval(lua_script, 1, lock_key, lock_value)
 

三、基础方案的三个坑

坑 1:业务跑太久,锁到期了(最常见)

 
时间线:
 
  T0: 你加锁成功,过期时间 30 秒
 
  T10: 你的 RAG 文档处理跑了一半
 
  T30: 锁到期自动释放
 
  T31: 另一个进程抢到了锁,也处理同一个文档
 
  T50: 第一个进程跑完了 → unlock
 
        → 但锁现在是别人的 → 把别人的锁删了 ❌
 
解法:看门狗自动续期(下面会讲)
 

坑 2:忘了设过期时间

 
# ❌ 错误
 
r.set(lock_key, lock_value, nx=True)  # 没设 ex!
 
# ✅ 正确:永远带上 ex
 
r.set(lock_key, lock_value, nx=True, ex=30)
 
# 即使进程挂了,30 秒后锁自动释放
 

坑 3:解锁没检查身份

 
# ❌ 错误
 
r.delete(lock_key)  # 不管谁的锁,直接删
 
# ✅ 正确:先检查再删(Lua 脚本)
 

四、看门狗(WatchDog)—— 自动续期

这是个啥

你用 Redis SET NX EX 加锁时设了个过期时间(比如 30 秒)。但你的业务可能跑 1 分钟。到期了锁没了,别人就能进来捣乱。

看门狗 = 一个后台任务,每过一段时间检查你的锁还在不在。在的话就给它续个期。


sequenceDiagram

    participant Client as 客户端

    participant WatchDog as 看门狗

    participant Redis as Redis

    Client->>Redis: SET NX EX 30(加锁)

    Note over Client: 业务执行中(跑了 60 秒)

    loop 每 10 秒续一次

        WatchDog->>Redis: PEXPIRE 续期 30 秒

        Redis-->>WatchDog: OK

    end

    Note over Client: [切片] [向量化] [存 ChromaDB] [更新索引] [完成]

    Client->>Client: finally unlock → 停止看门狗

Python 没有 Redisson —— 自己搓一个

你用 Python + FastAPI,没有 Redisson(那是 Java 的)。但原理一样可以自己实现:

 
import asyncio
 
import uuid
 
from redis.asyncio import Redis
 
class RedisLock:
 
    def __init__(self, redis: Redis, lock_key: str, timeout: int = 30):
 
        self.redis = redis
 
        self.lock_key = lock_key
 
        self.lock_value = str(uuid.uuid4())
 
        self.timeout = timeout  # 锁过期时间(秒)
 
        self._renew_task = None  # 看门狗任务
 
    async def acquire(self) -> bool:
 
        """加锁"""
 
        locked = await self.redis.set(
 
            self.lock_key, self.lock_value,
 
            nx=True, ex=self.timeout
 
        )
 
        if locked:
 
            self._start_watchdog()
 
        return locked
 
    def _start_watchdog(self):
 
        """启动看门狗:每 timeout/3 秒续一次"""
 
        async def _watchdog():
 
            while True:
 
                await asyncio.sleep(self.timeout / 3)  # 每 10 秒
 
                # 续期:把过期时间重置为 timeout
 
                await self.redis.pexpire(
 
                    self.lock_key, self.timeout * 1000
 
                )
 
        self._renew_task = asyncio.create_task(_watchdog())
 
    async def release(self):
 
        """释放锁"""
 
        if self._renew_task:
 
            self._renew_task.cancel()  # 停掉看门狗
 
            self._renew_task = None
 
        # Lua 脚本解锁(还是那套)
 
        lua = """
 
        if redis.call('get', KEYS[1]) == ARGV[1] then
 
            return redis.call('del', KEYS[1])
 
        else
 
            return 0
 
        end
 
        """
 
        await self.redis.eval(lua, 1, self.lock_key, self.lock_value)
 

用的时候:

 
from redis.asyncio import Redis
 
async def process_document(doc_id: str):
 
    r = Redis(decode_responses=True)
 
    lock = RedisLock(r, f"lock:doc:{doc_id}")
 
    if not await lock.acquire():
 
        return {"error": "该文档正在处理中,请稍后"}
 
    try:
 
        # RAG 文档处理(可能要跑几十秒)
 
        await chunk_document(doc_id)
 
        await embed_chunks(doc_id)
 
        await store_to_chromadb(doc_id)
 
        return {"status": "ok"}
 
    finally:
 
        await lock.release()
 

看门狗本质 = asyncio.create_task 起一个循环续期的后台协程,执行完毕就 cancel 掉。


五、写个 FastAPI 里能直接用的看门狗锁

 
# app/utils/lock.py
 
import asyncio
 
import uuid
 
from contextlib import asynccontextmanager
 
from redis.asyncio import Redis
 
@asynccontextmanager
 
async def distributed_lock(redis: Redis, lock_key: str, timeout: int = 30):
 
    """
 
    分布式锁上下文管理器(自带看门狗)
 
    用法:
 
        async with distributed_lock(redis, "lock:doc:123") as acquired:
 
            if not acquired:
 
                return {"error": "资源被占用"}
 
            # 执行业务...
 
    """
 
    lock_value = str(uuid.uuid4())
 
    locked = await redis.set(lock_key, lock_value, nx=True, ex=timeout)
 
    if not locked:
 
        yield False
 
        return
 
    # 启动看门狗
 
    watchdog_task = None
 
    async def watchdog():
 
        while True:
 
            await asyncio.sleep(timeout / 3)
 
            await redis.pexpire(lock_key, timeout * 1000)
 
    watchdog_task = asyncio.create_task(watchdog())
 
    try:
 
        yield True
 
    finally:
 
        if watchdog_task:
 
            watchdog_task.cancel()
 
        # Lua 解锁
 
        lua = """
 
        if redis.call('get', KEYS[1]) == ARGV[1] then
 
            return redis.call('del', KEYS[1])
 
        else
 
            return 0
 
        end
 
        """
 
        await redis.eval(lua, 1, lock_key, lock_value)
 

业务代码里用:

 
from app.utils.lock import distributed_lock
 
@router.post("/documents/{doc_id}/process")
 
async def process_document(doc_id: str):
 
    async with distributed_lock(redis, f"lock:doc:{doc_id}") as acquired:
 
        if not acquired:
 
            raise HTTPException(423, "文档正在处理中")
 
        # ... 实际处理逻辑
 
        return {"status": "ok"}
 

六、Redis 主从切换 —— 锁丢了怎么办

先搞懂 Redis 主从是怎么工作的

 
你有一个主库(Master)和一个从库(Slave)。
 
所有写操作都写 Master,Master 再把数据同步给 Slave。
 
  写锁 → Master    ───同步──→  Slave
 
        └─ "lock:doc:1001" 存在   └─ 还没收到 ❌
 
问题就在这里:同步是需要时间的(毫秒级)
 

丢锁到底是怎么发生的?逐步拆解

 
第 1 步:正常的分布式锁写入
 
  进程 A → Redis 主库(Master)
 
  进程 A:SET lock:doc:1001 "uuid-A" NX EX 30
 
  主库:OK,锁记录已保存 ✅
 
第 2 步:锁还没来得及同步到从库
 
  主库内部:
 
    内存里:lock:doc:1001 = "uuid-A" ✅
 
    准备同步给从库:还没发出去
 
    从库:不知道有这把锁
 
第 3 步:主库宕机了
 
  啪!主库挂了(停电/进程崩溃/网络不通)
 
  主库内存里的 lock:doc:1001 丢了
 
第 4 步:从库升级为新的主库
 
  Redis Sentinel 检测到主库挂了
 
  把从库提升为新的主库
 
  问题:新主库的内存里没有 lock:doc:1001
 
第 5 步:另一个进程来加锁
 
  进程 B → 新主库
 
  进程 B:SET lock:doc:1001 "uuid-B" NX EX 30
 
  新主库:lock:doc:1001 不存在 → 可以设置 → OK ✅
 
  结果:
 
    进程 A 以为自己持有锁(真的在干活)
 
    进程 B 也以为自己持有锁(也在干活)
 
    → 两个进程同时处理同一个资源 → 并发冲突 ❌
 

总结成一句话:锁写到 A 机器,还没同步到 B 机器,A 挂了。B 上位后不知道有这把锁,别人就能再抢到。


一句话人类版本

 
你在前台(主库)登记了"会议室 101 被 A 占了"。
 
前台还没把这个记录传到保安室(从库),
 
前台突然晕了(宕机)。
 
保安(从库)变成新的前台。
 
B 来问:"会议室 101 有人用吗?"
 
新前台说:"没人登记,你用吧。"
 
于是 A 和 B 同时进了同一间会议室。
 

这问题严重吗?—— 看场景

 
场景 A:扣钱 / 转账 / 库存扣减
 
  A 和 B 同时减库存 → 库存变成负数 → 超卖 ❌
 
  → 主从丢锁很严重
 
场景 B:RAG 文档处理
 
  A 在处理一个文档切片,B 也在处理同一个文档
 
  最多重复存两份一样的切片向量数据
 
  → 丢几条重复数据,不影响大局 🟡
 
场景 C:定时任务调度
 
  凌晨 3 点的归档任务,A 在跑,B 也在跑
 
  归档两次 → 多花一倍时间,结果一样 ✅
 
  → 基本没影响
 

所以:主从丢锁的问题严重程度 = 业务的容错程度。


解法有哪几种

解法 1:RedLock —— 多个 Redis 实例同时加锁

 
不用 1 个 Redis,用 5 个互相独立的 Redis(不存在主从关系)。
 
加锁流程:
 
  进程 A 同时向 5 个 Redis 发 SET NX EX
 
  只要超过半数(>=3 个)返回了 OK → 加锁成功
 
优点:
 
  其中一两个 Redis 挂了也不影响,别的还有锁记录
 
缺点:
 
  需要部署 5 个独立的 Redis 实例,成本高
 
  加锁要请求 5 次,慢
 
  Python 里实现起来比较麻烦
 
  业界有争议(DDIA 作者专门写文章怼过)
 

解法 2:WAIT 命令确认从库同步完成

 
sleep 的问题是"赌",WAIT 是"等确认"。
 
Redis 有个 WAIT 命令:
 
  WAIT <副本数> <超时毫秒>
 
加锁后调用 WAIT,阻塞到从库确认收到数据才继续:
 
  await redis.set(lock_key, uuid, nx=True, ex=30)
 
  replicas = await redis.wait(1, 5000)
 
  # 等待至少 1 个从库确认,最多等 5 秒
 
  # 返回实际确认的从库数量
 
  if replicas < 1:
 
      # 没有从库确认 → 保险起见,删锁,放弃
 
      await redis.delete(lock_key)
 
      raise Exception("主从同步失败,无法保证锁的可靠性")
 
  try:
 
      await do_work()
 
  finally:
 
      await unlock()
 
优点:
 
  └─ 真的能保证从库有锁记录了,不是赌
 
缺点:
 
  └─ 从库慢的话要等,加锁延迟变大
 
  └─ 如果只有一个 Redis 节点(没从库),WAIT 直接返回 0
 

解法 3:换成 ZooKeeper / etcd

 
ZooKeeper 用的是 ZAB 协议(强一致性):
 
  写请求必须多数节点确认后才算成功
 
  不存在"写了主库但还没同步"的情况
 
  主挂了,选举出来的新节点数据是完整的
 
优点:锁不会丢,保证互斥
 
缺点:ZK 部署比 Redis 复杂,性能也差一些
 
      如果项目没有 ZK,为了一个锁专门部署 ZK 太夸张
 

实际生产中最常见的做法

不是追求”锁一定不丢”,而是让业务能容忍”锁丢了”:

 
让业务支持幂等(idempotent)。
 
幂等 = 同一个操作执行 1 次和执行 N 次,结果一样。
 
生成订单 -> 不是幂等的(下两次单就多两个订单)
 
查询订单 -> 是幂等的(查多少次都是同一个结果)
 
...
 
怎么让操作变成幂等的?
 
  每个请求带一个唯一 ID(request_id/idempotent_key)
 
  处理之前先查这个 ID 有没有处理过
 
  处理过就直接返回结果,不再重复执行
 
例子:
 
  POST /api/orders
 
  Header: Idempotency-Key: uuid-xxxx
 
  服务端处理:
 
    查 Redis(或 DB):这个 key 处理过吗?
 
      处理过 → 返回之前的结果 ✅
 
      没处理过 → 执行下单,保存结果到 Redis
 
  就算分布式锁丢了,两个请求同时进来,
 
  因为都有相同的 Idempotency-Key,
 
  只有第一个真正处理,第二个直接返回缓存结果。
 

锁防并发,幂等防重复。锁可能失效,幂等是最后一道防线。


一句话讲清

 
延伸提问:"Redis 主从切换会导致锁丢失吗?"
 
你:
 
"会的。在极端情况下会丢锁——
 
锁写到主库,还没同步到从库,主库挂了,
 
从库上位后没有这把锁的记录,
 
别的进程就能加锁成功。
 
但我们项目对这个问题做了权衡:
 
一般业务场景用看门狗续期 + 业务幂等就够了。
 
像 RAG 文档处理这种场景,
 
两个进程同时处理同一个文档,最多多存一份向量数据,
 
对用户没影响,并不需要为了完美的一致性
 
去上 RedLock 或 ZK 那么复杂的方案。
 
如果是扣库存、扣钱这种场景,
 
那就需要用 Idempotency-Key 做幂等兜底,
 
或者直接用分布式事务框架了。"
 

七、话术(针对你方向的改编版)

 
延伸提问:"分布式锁怎么实现?"
 
你:
 
"我项目里用的是 Redis 分布式锁。
 
最基础的是 SET key value NX EX 一条命令:
 
  NX 保证互斥(key 不存在才设),EX 保证防死锁(自动过期)。
 
value 必须用 UUID 标识持有者。
 
解锁用 Lua 脚本对比 value 再 DEL,防止误删别人的锁。
 
两条命令之间有间隙,必须用 Lua 原子执行。
 
这个是 Python 实现的核心代码(理解即可,能口述更好):
 
  locked = redis.set(key, uuid, nx=True, ex=30)
 
  if locked:
 
      try:
 
          process()
 
      finally:
 
          redis.eval(lua_script, 1, key, uuid)
 
延伸提问:"如果业务执行超过 30 秒呢?"
 
你:
 
"所以我加了看门狗机制——本质是一个后台协程,
 
每隔 timeout/3 秒给锁续一次期。
 
业务没跑完,锁就不会到期。
 
执行完 finally 里 unlock 时,顺手 cancel 掉看门狗。
 
实际项目里我把它封装成了 async with 上下文管理器,
 
用起来就一行。"
 

记忆口诀

SET NX EX 一条命令——原子加锁+过期,防死锁。

UUID 标身份,Lua 验身再删——防误删别人的锁。

看门狗=后台协程每 10 秒续一次——业务不结束,锁不释放。

主从切换可能丢锁——业务幂等比锁的绝对可靠更实用。

速记卡(面试闪卡)

Q1:一句话讲清「分布式锁:SET NX EX + 看门狗」到底是什么?

A:分布式锁让多台机器抢同一把全局锁,Redis 用 SET NX EX 一条命令原子加锁防死锁。

Q2:一、SET NX EX —— 最基础的分布式锁 —— 怎么理解?

A:像抢唯一制冰机:SET lock NX(不存在才设,互斥)EX 30(30秒自动消失,防死锁)再写上 value(谁拿的,身份)。必须一条命令——分两条命令中间 Redis 宕机就永久死锁。NX(Not eXists,不存在才设)/ EX(过期秒数)。

Q3:二、Python 手写实现(必手写) —— 怎么理解?

A:像自己配钥匙:redis.set(nx=True, ex=30) 加锁,抢不到就报错重试。解锁不能 Python 里 if+delete(两步之间锁可能过期误删别人的),必须用 Lua 脚本在 Redis 端原子”验身份再删”。Lua(Lua 脚本,服务端原子执行)。

Q4:三、基础方案的三个坑 —— 怎么理解?

A:像三个陷阱:① 业务跑太久锁过期,别人进来你却删了别人的锁;② 忘了设 EX,进程崩了锁永不放→死锁;③ 解锁用 value 校验,不用 Lua 会误删。坑都源于”加锁和解锁不是原子”。

Q5:四、看门狗(WatchDog)—— 自动续期 —— 怎么理解?

A:像雇个保姆定时续租:业务跑 1 分钟但锁 30 秒到期,看门狗每 10 秒 PEXPIRE 续一次,业务跑完才停。Python 没 Redisson 就自己起后台任务搓一个。WatchDog(看门狗,自动续期机制)专治”锁提前过期”。

Q6:核心速记主线有哪些?

  • 加锁:SET key value NX EX,一条命令保证原子

  • 解锁:Lua 脚本验 value 再删,防误删别人锁

  • 三大坑:业务超时、忘设 EX、误删他人锁

  • 看门狗:后台定时续期,业务没完锁不过期

口诀

A:分布式锁抢全局,SET NX EX 一条令;

分两命令会死锁,原子才是真本领;

解锁必用 Lua 脚本,验明正身再删净;

看门狗来续租期,业务没完锁不醒。

相关链接