Python 线程同步

当多个线程并发地访问共享资源时，同步它们的访问对于保持数据完整性和程序正确性至关重要。Python 中的线程同步可以通过 threading 模块提供的多种同步原语来实现，例如锁、条件变量、信号量以及屏障等，以控制对共享资源的访问和协调多个线程的执行。

在本教程中，我们将学习 Python 的 threading 模块提供的各种同步原语。

使用锁进行线程同步

Python 的 threading 模块中的锁对象提供最简单的同步原语。它们允许线程在代码的关键部分获取和释放锁，确保一次只有一个线程可以执行受保护的代码。

通过调用 Lock() 方法创建一个新的锁，这将返回一个锁对象。可以使用 acquire(blocking) 方法获取锁，强制线程同步运行。可选的 blocking 参数允许你控制线程是否等待获取锁，并使用 release() 方法释放锁。

示例

以下示例演示了如何使用锁（threading.Lock() 方法）在 Python 中同步线程，确保多个线程安全正确地访问共享资源。

import threading

counter = 10

def increment(theLock, N):
   global counter
   for i in range(N):
      theLock.acquire()
      counter += 1
      theLock.release()

lock = threading.Lock()
t1 = threading.Thread(target=increment, args=[lock, 2])
t2 = threading.Thread(target=increment, args=[lock, 10])
t3 = threading.Thread(target=increment, args=[lock, 4])

t1.start()
t2.start()
t3.start()

# 等待所有线程完成
for thread in (t1, t2, t3):
   thread.join()

print("所有线程已完成")
print("最终计数器值:", counter)

输出：

所有线程已完成
最终计数器值: 26

使用条件变量同步 Python 线程

条件变量使线程能够在被另一个线程通知之前等待。它们对于在线程之间提供通信很有用。wait() 方法用于阻塞线程直到它被另一个线程通过 notify() 或 notify_all() 通知。

示例

此示例演示了如何使用 notify() 和 wait() 方法通过条件变量同步线程。

import threading

counter = 0  

# 消费者函数
def consumer(cv):
   global counter
   with cv:
      print("消费者正在等待")
      cv.wait()  # 等待直到被 increment 通知
      print("消费者已被通知。当前计数器值:", counter)

# increment 函数
def increment(cv, N):
   global counter
   with cv:
      print("increment 正在生产项目")
      for i in range(1, N + 1):
         counter += i  # 计数器递增 i
        
      # 通知消费者 
      cv.notify()  
      print("Increment 已完成")

# 创建一个条件对象
cv = threading.Condition()

# 创建并启动线程
consumer_thread = threading.Thread(target=consumer, args=[cv])
increment_thread = threading.Thread(target=increment, args=[cv, 5])

consumer_thread.start()
increment_thread.start()

consumer_thread.join()
increment_thread.join()

print("最终计数器值:", counter)

输出：

消费者正在等待
increment 正在生产项目
Increment 已完成
消费者已被通知。当前计数器值: 15
最终计数器值: 15

使用 join() 方法同步线程

Python 的 threading 模块中的 join() 方法用于等待所有线程完成它们的执行。这是一种简单的方法来同步主线程和其他线程的完成。

示例

此示例演示了如何使用 join() 方法来确保主线程等待所有启动的线程完成它们的工作然后再继续。

import threading
import time

class MyThread(threading.Thread):
   def __init__(self, threadID, name, counter):
      threading.Thread.__init__(self)
      self.threadID = threadID
      self.name = name
      self.counter = counter
      
   def run(self):
      print("开始 " + self.name)    
      print_time(self.name, self.counter, 3)
      
def print_time(threadName, delay, counter):
   while counter:
      time.sleep(delay)
      print("%s: %s" % (threadName, time.ctime(time.time())))
      counter -= 1
      
threads = []

# 创建新线程
thread1 = MyThread(1, "Thread-1", 1)
thread2 = MyThread(2, "Thread-2", 2)

# 启动新线程
thread1.start()
thread2.start()

# 合并线程
thread1.join()
thread2.join()

print("退出主线程")

输出：

开始 Thread-1
开始 Thread-2
Thread-1: Mon Jul  1 16:05:14 2024
Thread-2: Mon Jul  1 16:05:15 2024
Thread-1: Mon Jul  1 16:05:15 2024
Thread-1: Mon Jul  1 16:05:16 2024
Thread-2: Mon Jul  1 16:05:17 2024
Thread-2: Mon Jul  1 16:05:19 2024
退出主线程

额外的同步原语

除了上述同步原语之外，Python 的 threading 模块还提供了：

RLocks（重入锁）：锁的一种变体，允许一个线程多次获取同一个锁，这对于递归函数或嵌套函数调用很有用。
Semaphores（信号量）：类似于锁，但是带有一个计数器。线程可以获取信号量直到初始化时定义的某个限度。信号量对于限制固定容量资源的访问很有用。
Barriers（屏障）：允许固定数量的线程在屏障点同步，并且只有当所有线程都到达该点时才能继续执行。屏障对于协调必须全部完成某个阶段的执行才能进一步前进的一组线程很有用。