Python确保函数A单实例运行：Lock锁失效问题求助

核心问题在于你之前使用的Lock作用域不对——如果锁是每个实例私有或者在方法内临时创建的，多个请求会使用不同的锁，根本起不到互斥效果。必须让所有调用A的请求共享同一个全局/类级别的锁。

下面是两种可靠的实现方案：

方案一：类级别的Lock（推荐） #

给First类定义一个类属性lock，所有实例共享这个锁，确保同一时间只有一个A方法执行：

import threading

class First:
    # 类级别的锁，所有First实例共用
    lock = threading.Lock()

    def A(self, id):
        # 使用with语句自动处理锁的获取和释放，避免忘记解锁
        with self.lock:
            print(f"开始执行A，请求ID: {id}")
            update_db_table(id + 1)
            # 模拟数据库操作耗时，方便测试效果
            # time.sleep(1)

class Second:
    def B(self, id):
        first = First()
        first.A(id)

方案二：类方法+类级锁 #

如果A不需要操作实例属性，可以改成类方法，直接通过类调用，锁的共享逻辑更清晰：

import threading

class First:
    lock = threading.Lock()

    @classmethod
    def A(cls, id):
        with cls.lock:
            print(f"开始执行A，请求ID: {id}")
            update_db_table(id + 1)

class Second:
    def B(self, id):
        First.A(id)

测试验证 #

用多线程模拟100个客户端请求，验证互斥效果：

import threading
import time

# 模拟数据库更新函数
def update_db_table(val):
    print(f"数据库更新完成，值: {val}")
    time.sleep(1)

# 上面的类定义...

def run_test():
    second = Second()
    threads = []
    # 模拟10个并发请求（可改成100）
    for i in range(10):
        t = threading.Thread(target=second.B, args=(i,))
        threads.append(t)
        t.start()
    
    # 等待所有线程完成
    for t in threads:
        t.join()

if __name__ == "__main__":
    run_test()

关键注意点 #

• 锁必须是全局共享的：类属性、全局变量都是可行的选择，绝对不能在A方法内部或者实例初始化时创建锁。
• 多进程场景需换锁：如果你的服务是多进程架构（比如Gunicorn的多进程模式），threading.Lock会失效，需要改用multiprocessing.Lock或者基于Redis的分布式锁。
• 异步场景用asyncio.Lock：如果是异步Web框架（比如FastAPI），要使用asyncio.Lock而非线程锁，避免阻塞事件循环。

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