Python 3.12及以上带冷却期的线程限频锁需求

需求回顾 #

• 具备标准线程锁的互斥特性，同一时间仅一个线程可持有锁
• 锁释放后，在指定冷却期（默认1秒）内无法被重新获取，用于速率限制
• 仅提供阻塞式acquire方法，无额外参数
• 实现与threading.Lock一致的release、上下文管理器接口（__enter__/__exit__）

无竞态条件的实现方案 #

以下是符合Python 3.12+要求的可靠实现，通过内部互斥锁保护状态，彻底避免竞态：

import threading
import time
from typing import Optional

class CooldownLock:
    def __init__(self, cooldown: float = 1):
        self._cooldown = cooldown
        self._inner_lock = threading.Lock()
        self._last_release_time: Optional[float] = None

    def acquire(self):
        # 先获取内部互斥锁，保证状态操作的原子性
        self._inner_lock.acquire()
        try:
            now = time.time()
            # 如果冷却期未过，等待剩余时间
            if self._last_release_time is not None:
                wait_until = self._last_release_time + self._cooldown
                if now < wait_until:
                    time.sleep(wait_until - now)
        finally:
            # 内部锁保持持有状态，直到release被调用
            pass

    def release(self):
        # 记录释放时间，然后释放内部锁
        self._last_release_time = time.time()
        self._inner_lock.release()

    def __enter__(self):
        self.acquire()
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        self.release()

实现说明 #

• 用threading.Lock作为内部核心锁，确保所有状态操作（检查冷却时间、记录释放时间）都是原子性的，彻底消除竞态风险
• acquire方法会先拿到内部锁，再检查是否需要等待冷却期，等待完成后保持内部锁持有，直到release被调用
• release方法仅更新释放时间并释放内部锁，完全不阻塞，符合要求
• 上下文管理器接口与标准threading.Lock完全兼容，可直接用在with语句中

使用示例 #

def worker(lock: CooldownLock, name: str):
    for _ in range(3):
        with lock:
            print(f"{name} 持有锁，当前时间: {time.time():.2f}")
            # 模拟业务操作
            time.sleep(0.5)

lock = CooldownLock(cooldown=1)
t1 = threading.Thread(target=worker, args=(lock, "线程1"))
t2 = threading.Thread(target=worker, args=(lock, "线程2"))

t1.start()
t2.start()
t1.join()
t2.join()

运行后可以看到，每个线程释放锁后，下一次获取锁会间隔至少1秒，同时保证同一时间只有一个线程持有锁。

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