Python3.12多线程中多共享变量的锁策略选型问题

核心结论 #

这三个变量（N、P、Limits）属于关联状态集合，修改时必须保证原子性，因此只需要一个锁就足够，完全不需要多个锁。多个锁反而可能引发死锁风险，且毫无必要。

你原有代码的关键错误 #

你的代码未达预期，核心问题出在以下几点：

1. 默认参数陷阱：def update_variables_1(n=N, p=P, limits=Limits) 中的默认参数是在函数定义时求值的，并非调用时的最新值；而且函数内修改的是局部变量n/p/limits，根本没改动全局的N/P/Limits。
2. 锁的位置错误：把函数定义放在with lock块里毫无意义，锁需要包裹的是修改共享变量的核心逻辑，而非函数定义本身。
3. Lock对象使用错误：你写的with Lock是直接引用类，应该先实例化一个锁对象，比如lock = threading.Lock()，再用with lock。

正确的实现方式 #

步骤1：定义全局锁与共享变量 #

导入threading模块，创建全局锁实例，同时定义需要共享的变量：

import threading
from math import pow

# 共享状态变量
N = 1
P = 2
Limits = 3.0

# 全局锁，所有操作共享变量的线程共用这一个锁
lock = threading.Lock()

步骤2：编写线程安全的更新函数 #

直接操作全局变量，并用锁包裹整个修改逻辑，确保原子性：

def update_variables_1():
    global N, P, Limits
    with lock:
        # 先计算所有新值，避免中间状态被其他线程读取
        new_n = N + pow(N, 2)
        new_p = P + new_n
        new_limits = (new_n + new_p) / 2
        # 一次性更新所有变量，保证状态完全一致
        N, P, Limits = new_n, new_p, new_limits

def update_variables_2():
    global N, P, Limits
    with lock:
        new_n = N - pow(N, 2)
        new_p = P - new_n
        new_limits = (new_n - new_p) / 2
        N, P, Limits = new_n, new_p, new_limits

步骤3：启动线程 #

在T2（生产者）和T3（日志线程）中调用上述函数即可：

def thread_t2():
    # 生产者线程逻辑，示例为循环调用更新函数
    while True:
        update_variables_1()
        # 可选：添加sleep避免过度占用CPU
        # time.sleep(0.1)

def thread_t3():
    # 日志线程逻辑，示例为循环调用更新函数
    while True:
        update_variables_2()
        # time.sleep(0.1)

# 启动线程
t2 = threading.Thread(target=thread_t2)
t3 = threading.Thread(target=thread_t3)
t2.start()
t3.start()

额外注意事项 #

• 如果T1消费者线程需要读取这三个变量，读取操作也必须加同一个锁，否则可能读到不一致的中间状态：

  def read_shared_vars():
      global N, P, Limits
      with lock:
          return (N, P, Limits)
  

• 更优雅的方式是把共享状态和锁封装到类中，避免全局变量：

  class SharedState:
      def __init__(self):
          self.N = 1
          self.P = 2
          self.Limits = 3.0
          self.lock = threading.Lock()
  
      def update_1(self):
          with self.lock:
              new_n = self.N + pow(self.N, 2)
              new_p = self.P + new_n
              new_limits = (new_n + new_p) / 2
              self.N, self.P, self.Limits = new_n, new_p, new_limits
  
      def update_2(self):
          with self.lock:
              new_n = self.N - pow(self.N, 2)
              new_p = self.P - new_n
              new_limits = (new_n - new_p) / 2
              self.N, self.P, self.Limits = new_n, new_p, new_limits
  
  # 使用方式
  state = SharedState()
  
  def thread_t2():
      while True:
          state.update_1()
  
  def thread_t3():
      while True:
          state.update_2()
  

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