Python 使用 functools 实现注解同步方法

Chloe

发布时间：2026-10-11| 3人浏览过

原创

Python 原生缺乏类似 Java 的 synchronized 关键字，通常需手动管理 threading.Lock()。本文介绍通过 functools 封装装饰器，将锁的获取与释放逻辑隐藏于注解中，实现优雅的方法级同步控制，避免手动加锁释放锁的代码冗余。

Python 实现方法级同步：利用 functools 装饰器封装 Lock

在 Java 等语言中，synchronized 关键字提供了便捷的线程同步机制。然而，Python 标准库并未提供同名的关键字。在多线程环境下，若需保证方法的原子性，开发者通常需手动实例化 threading.Lock() 并在方法内部显式调用 acquire() 和 release()。这种手动管理锁的方式不仅代码冗长，还容易因异常导致锁未正确释放，从而引发死锁或资源竞争问题。

为解决这一痛点，可以利用 Python 的 functools 模块结合装饰器模式，将锁的获取与释放逻辑封装在注解中。通过自定义 @synchronized 装饰器，开发者只需在目标方法前添加注解，即可实现类似 Java 的同步效果，使代码更加简洁且易于维护。

核心实现逻辑

实现该功能的关键在于定义一个高阶函数 synchronized，它接收原始方法作为参数，并返回一个包装后的方法 wrapper。在 wrapper 内部，利用 with self.lock: 语句块自动管理锁的生命周期：进入代码块时获取锁，退出时（无论正常返回还是抛出异常）自动释放锁。

具体实现步骤如下：

导入依赖库：引入 threading 用于创建锁和线程，functools 用于保留原函数的元数据（如名称和文档字符串）。
定义装饰器：创建 synchronized 函数，内部使用 @functools.wraps(func) 装饰 wrapper，确保被装饰方法的元信息不丢失。
封装锁逻辑：在 wrapper 中，通过 with self.lock: 获取锁，并调用原始方法 func(self, *args, **kwargs)。
类初始化：在包含同步方法的类中，于 __init__ 方法内初始化锁对象 self.lock = threading.Lock()。
应用注解：在需要同步的方法前添加 @synchronized 注解。

完整代码示例

以下代码展示了如何定义装饰器、创建包含同步方法的类，并通过多线程验证同步效果。代码中定义了三个测试方法 test1、test2 和 test3，它们均被 @synchronized 装饰。启动三个线程分别执行这些方法，由于共享同一把锁，输出结果将按顺序交替打印，而非交错混乱，证明了同步机制的有效性。

import threading
import functools
import time

def synchronized(func):
  @functools.wraps(func)
  def wrapper(self, *args, **kwargs):
    with self.lock:
      return func(self, *args, **kwargs)
  return wrapper

class MyWorker(object):
  def __init__(self):
    self.lock = threading.Lock()
    self.idx = 0
  
  @synchronized
  def test1(self):
    for i in range(1, 11):
      self.idx = self.idx + 1
      print "Test1: " + str(self.idx)
      time.sleep(1)
  
  @synchronized
  def test2(self):
    for i in range(1, 11):
      self.idx = self.idx + 1
      print "Test2: " + str(self.idx)
      time.sleep(1)
  
  @synchronized
  def test3(self):
    for i in range(1, 11):
      self.idx = self.idx + 1
      print "Test3: " + str(self.idx)
      time.sleep(1)

worker = MyWorker()
threading.Thread(target=worker.test1).start()
threading.Thread(target=worker.test2).start()
threading.Thread(target=worker.test3).start()