Python多线程同步机制详解：Lock、RLock、Semaphore、Event与Condition

在多线程应用开发中，如何确保线程安全、实现线程间同步以及正确处理共享变量访问，是极具挑战性的核心问题。若处理不当，可能导致严重的数据错误或程序崩溃。Python标准库提供了Lock、RLock、Semaphore、Event和Condition等同步工具，旨在解决线程间的同步协调以及共享变量访问的互斥保护问题。

Lock与RLock主要用于解决多线程访问共享变量时的互斥问题；Semaphore是Lock的增强版，允许同时被多个线程持有，而Lock仅允许单一线程持有；Event作为线程间通信机制，类似于信号量，允许一个线程发送信号触发另一个线程执行操作；Condition则用于在特定事件触发或满足特定条件后才处理数据，实现更复杂的同步逻辑。

1、Lock（互斥锁）

其工作流程通常包括：请求锁定、进入锁定池等待、获取锁、进入已锁定状态，最后释放锁。

Lock（指令锁）是最低级别的同步指令。当Lock处于锁定状态时，它不属于任何特定线程。Lock包含两种状态：锁定和非锁定，以及两个基本操作方法。

可以将Lock想象为一个锁定池。当线程请求锁定时，该线程会被放入池中，直到获得锁定后才会出池。池中的线程处于同步阻塞状态。

构造方法： Lock()

实例方法： acquire([timeout])：使线程进入同步阻塞状态，尝试获取锁定。 release()：释放锁。调用前线程必须已获得锁定，否则将抛出异常。

if mutex.acquire():
 counter += 1
 print "I am %s, set counter:%s" % (self.name, counter)
  mutex.release()

2、RLock（可重入锁）

RLock（可重入锁）是一种允许同一线程多次请求的同步指令。RLock引入了“拥有线程”和“递归等级”的概念。当处于锁定状态时，RLock被某个特定线程拥有。拥有RLock的线程可以再次调用acquire()，但释放锁时需要调用相同次数的release()。

可以认为RLock包含一个锁定池和一个初始值为0的计数器。每次成功调用acquire()或release()时，计数器分别+1或-1。当计数器为0时，锁处于未锁定状态。

构造方法： RLock()

实例方法： acquire([timeout])/release()：功能与Lock类似。

3、Semaphore（共享对象访问）

接下来讨论Semaphore。实际上，Semaphore是我较晚使用的同步锁，以前类似的实现通常用RLock完成，相对而言较为复杂，因为RLock需要成对的锁定和解锁操作。

Semaphore管理一个内置计数器： 每当调用acquire()时，内置计数器减1； 调用release()时，内置计数器加1； 计数器不能小于0。当计数器为0时，acquire()将阻塞线程，直到其他线程调用release()。

直接展示代码，我们将semaphore控制为3，意味着同时只有3个线程可以使用该锁，其余线程只能阻塞等待。

#coding:utf-8
#blog xiaorui.cc
import time
import threading

semaphore = threading.Semaphore(3)

def func():
 if semaphore.acquire():
  for i in range(3):
   time.sleep(1)
   print (threading.currentThread().getName() + '获取锁')
  semaphore.release()
  print (threading.currentThread().getName() + ' 释放锁')


for i in range(5):
 t1 = threading.Thread(target=func)
 t1.start()

4、Event（线程间通信）

Event内部包含一个标志位，初始值为false。 可以使用set()将其设置为true； 或者使用clear()将其重新设置为false； 可以使用is_set()来检查标志位的状态。

另一个最重要的函数是wait(timeout=None)，用于阻塞当前线程，直到Event的内部标志位被设置为true或timeout超时。如果内部标志位为true，则wait()函数立即返回。

import threading
import time

class MyThread(threading.Thread):
 def __init__(self, signal):
  threading.Thread.__init__(self)
  self.singal = signal

 def run(self):
  print "I am %s,I will sleep ..."%self.name
  self.singal.wait()
  print "I am %s, I awake..." %self.name

if __name__ == "__main__":
 singal = threading.Event()
 for t in range(0, 3):
  thread = MyThread(singal)
  thread.start()

 print "main thread sleep 3 seconds... "
 time.sleep(3)

 singal.set()

5、Condition（线程同步）

可以将Condition理解为一把高级锁，它提供了比Lock和RLock更高级的功能，允许控制复杂的线程同步问题。threading.Condition在内部维护一个锁对象（默认是RLock），可以在创建Condition对象时传入锁对象作为参数。Condition也提供了acquire和release方法，其含义与锁的acquire和release方法一致，实际上只是简单调用内部锁对象的对应方法。Condition还提供了以下方法（特别注意：这些方法只有在占用锁（acquire）之后才能调用，否则将报RuntimeError异常）：

Condition.wait([timeout]):

wait方法释放内部占用的锁，同时线程被挂起，直至接收到通知被唤醒或超时（如果提供了timeout参数）。当线程被唤醒并重新占有锁时，程序才会继续执行。

Condition.notify():

唤醒一个挂起的线程（如果存在挂起的线程）。注意：notify()方法不会释放所占用的锁。

Condition.notify_all() Condition.notifyAll()

唤醒所有挂起的线程（如果存在挂起的线程）。注意：这些方法不会释放所占用的锁。

关于Condition，这里有一个示例供参考。

from threading import Thread, Condition
import time
import random

queue = []
MAX_NUM = 10
condition = Condition()

class ProducerThread(Thread):
 def run(self):
  nums = range(5)
  global queue
  while True:
   condition.acquire()
   if len(queue) == MAX_NUM:
    print "Queue full, producer is waiting"
    condition.wait()
    print "Space in queue, Consumer notified the producer"
   num = random.choice(nums)
   queue.append(num)
   print "Produced", num
   condition.notify()
   condition.release()
   time.sleep(random.random())


class ConsumerThread(Thread):
 def run(self):
  global queue
  while True:
   condition.acquire()
   if not queue:
    print "Nothing in queue, consumer is waiting"
    condition.wait()
    print "Producer added something to queue and notified the consumer"
   num = queue.pop(0)
   print "Consumed", num
   condition.notify()
   condition.release()
   time.sleep(random.random())


ProducerThread().start()
ConsumerThread().start()

以上就是本文的全部内容，希望对大家的学习有所帮助，也希望大家多多支持本站。