一文速通Python并行计算：03 Python多线程编程-多线程同步（上）—基于互斥 ...

钜形不锈钢水箱 · 2025-3-26 22:11:32

一文速通 Python 并行计算：03 Python 多线程编程-多线程同步（上）—基于互斥锁、递归锁和信号量

摘要：

在 Python 多线程编程中，线程同步是确保多个线程安全访问共享资源的关键技能。本篇文章先容了互斥锁（Lock）、递归锁（RLock） 和 信号量（Semaphore） 的概念与应用。互斥锁用于防止多个线程同时修改数据，递归锁适用于嵌套锁定场景，而信号量则限制同时访问资源的线程数。

关于我们更多先容可以检察云文档： Freak 嵌入式工作室云文档 ，大概访问我们的 wiki： https://github.com/leezisheng/Doc/wik

正文

1.线程同步的基本概念

在 Python 多线程编程中，线程同步（Thread Synchronization）是确保多个线程安全地访问共享资源的机制，在多线程环境下，如果多个线程同时访问和修改共享资源，可能会导致数据竞争（Race Condition）和数据不一致性（Data Inconsistency）题目。例如：

import threading
# 共享变量
counter = 0
def increment():
global counter
for _ in range(1000000):
counter += 1
# 创建两个线程
t1 = threading.Thread(target=increment)
t2 = threading.Thread(target=increment)
# 启动线程
t1.start()
t2.start()
# 等待线程执行完毕
t1.join()
t2.join()
# 预期是 2000000，但可能小于这个值
print("Final Counter:", counter)

复制代码

由于 counter += 1 并不是原子操作，而是 读取 -> 计算 -> 写入 三步操作，因此两个线程可能同时读取 counter，导致写入时丢失部分数据，最终的结果可能小于 2000000，这就是 竞争条件（Race Condition）。
再举一个实例，下面的代码我们了跑 200 个线程，但是这 200 个线程都会去访问 counter 这个公共资源，并对该资源进行处理(counter += 1)，我们看下运行结果：

import threading
import time
counter = 0
class MyThread(threading.Thread):
def __init__(self):
threading.Thread.__init__(self)
def run(self):
global counter
time.sleep(1);
counter += 1
print("I am %s, set counter:%s" % (self.name, counter))
if __name__ == "__main__":
for i in range(0, 200):
my_thread = MyThread()
my_thread.start()

复制代码

从中我们已经看出了这个全局资源(counter)被抢占的环境，题目产生的原因就是没有控制多个线程对同一资源的访问，对数据造成破坏，使得线程运行的结果不可预期。这种现象称为“线程不安全”。在开发过程中我们必须要避免这种环境。
2.基于互斥锁的线程数据同步

“线程不安全”最简朴的解决方法是使用锁。锁的操作非常简朴，当一个线程必要访问部分共享内存时，它必须先获得锁才能访问。此线程对这部分共享资源使用完成之后，该线程必须释放锁，然后其他线程就可以拿到这个锁并访问这部分资源了。
在 Python 中我们使用 threading 模块提供的 Lock 类来实现互斥锁的机制。我们对上面的程序进行整改，为此我们必要添加一个锁变量 mutex`` = threading.Lock()，然后在争夺资源的时间之前我们会先抢占这把锁 mutex``.acquire``()，对资源使用完成之后我们在释放这把锁 mutex``.release``()。
代码如下：

import threading
import time
counter = 0
mutex = threading.Lock()
class MyThread(threading.Thread):
def __init__(self):
threading.Thread.__init__(self)
def run(self):
global counter, mutex
time.sleep(1);
if mutex.acquire():
counter += 1
print("I am %s, set counter:%s" % (self.name, counter))
mutex.release()
if __name__ == "__main__":
for i in range(0, 100):
my_thread = MyThread()
my_thread.start()

复制代码

下图为输出，可以看到 counter 变量不断递增，全局资源(counter)被抢占的环境得到解决。主要过程为：当一个线程调用 Lock 对象的 acquire() 方法获得锁时，这把锁就进入 “locked” 状态。因为每次只有一个线程 1 可以获得锁，以是如果此时另一个线程 2 试图获得这个锁，该线程 2 就会变为 “block“ 同步壅闭状态。直到拥有锁的线程 1 调用锁的 release() 方法释放锁之后，该锁进入 “unlocked” 状态。线程调度程序从处于同步壅闭状态的线程中选择一个来获得锁，并使得该线程进入运行（running）状态。

然而，在实际使用的过程中，我们发现这个方法经常会导致一种糟糕的死锁现象。当差别的线程要求得到一个锁时，死锁就会发生，这时程序不可能继续实行，因为它们互相拿着对方必要的锁。

为了简化题目，我们设有两个并发的线程（线程 A 和线程 B )，必要资源 1 和资源 2 .假设线程 A 必要资源 1 ，线程 B 必要资源 2 .在这种环境下，两个线程都使用各自的锁，现在为止没有冲突。现在假设，在两边释放锁之前，线程 A 必要资源 2 的锁，线程 B 必要资源 1 的锁，没有资源线程不会继续实行。鉴于现在两个资源的锁都是被占用的，而且在对方的锁释放之前都处于等候且不释放锁的状态。这是死锁的典范环境。
我们来看如下代码：

import threading
counterA = 0
counterB = 0
mutexA = threading.Lock()
mutexB = threading.Lock()
class MyThread(threading.Thread):
def __init__(self):
threading.Thread.__init__(self)
def run(self):
self.fun1()
self.fun2()
def fun1(self):
global mutexA, mutexB
if mutexA.acquire():
print("I am %s , get res: %s" % (self.name, "ResA"))
if mutexB.acquire():
print("I am %s , get res: %s" % (self.name, "ResB"))
mutexB.release()
mutexA.release()
def fun2(self):
global mutexA, mutexB
if mutexB.acquire():
print("I am %s , get res: %s" % (self.name, "ResB"))
if mutexA.acquire():
print("I am %s , get res: %s" % (self.name, "ResA"))
mutexA.release()
mutexB.release()
if __name__ == "__main__":
for i in range(0, 100):
my_thread = MyThread()
my_thread.start()

复制代码

代码中展示了一个线程的两个功能函数分别在获取了一个竞争资源之后再次获取另外的竞争资源，我们看运行结果：

可以看到，程序已经挂起在那儿了，这种现象我们就称之为”死锁“。避免死锁主要方法就是：正确有序的分配资源，避免死锁算法中最有代表性的算法是 Dijkstra E.W 于 1968 年提出的银行家算法。
3.基于递归锁的线程数据同步

考虑这种环境：如果一个线程遇到锁嵌套的环境该怎么办，这个嵌套是指当我一个线程在获取临界资源时，又必要再次获取。代码如下：

import threading
import time
counter = 0
mutex = threading.Lock()
class MyThread(threading.Thread):
def __init__(self):
threading.Thread.__init__(self)
def run(self):
global counter, mutex
time.sleep(1);
if mutex.acquire():
counter += 1
print("I am %s, set counter:%s" % (self.name, counter))
if mutex.acquire():
counter += 1
print("I am %s, set counter:%s" % (self.name, counter))
mutex.release()
mutex.release()
if __name__ == "__main__":
for i in range(0, 200):
my_thread = MyThread()
my_thread.start()

复制代码

这种环境的代码运行环境如下，可以看到线程获取一次互斥锁后，必要再次使用临界变量 counter，于是再次实验获取互斥锁：

之后就直接挂起了，这种环境形成了最简朴的死锁。
那有没有一种环境可以在某一个线程使用互斥锁访问某一个竞争资源时，可以再次获取呢？在 Python 中为了支持在同一线程中多次请求同一资源，python 提供了“可重入锁”：threading.RLock。这个 RLock 内部维护着一个 Lock 和一个 counter 变量，counter 记录了 acquire 的次数，从而使得资源可以被多次 require。直到一个线程所有的 acquire 都被 release，其他的线程才能获得资源。上面的例子如果使用 RLock 取代 Lock，则不会发生死锁：
代码只需将上述的：

mutex = threading.Lock()

复制代码

替换成：

mutex = threading.RLock()

复制代码

这种锁对比 Lock 有是三个特点：

谁拿到谁释放。如果线程 A 拿到锁，线程 B 无法释放这个锁，只有 A 可以释放；
同一线程可以多次拿到该锁，即可以 acquire 多次；
acquire 多少次就必须 release 多少次，只有最后一次 release 才能改变 RLock 的状态为 unlocked。

嵌套锁也有缺点，它给我们的锁检测带来了麻烦。
4.基于信号量的线程数据同步

信号量由 E.Dijkstra 发明并第一次应用在操作系统中，信号量是由操作系统管理的一种抽象数据范例，用于在多线程中同步对共享资源的使用。本质上说，信号量是一个内部数据，用于标明当前的共享资源可以有多少并发读取。
同样的，在 threading 模块中，信号量的操作有两个函数 acquire() 和 release() ，解释如下：
（1）每当线程想要读取关联了信号量的共享资源时，必须调用 acquire() ，此操作减少信号量的内部变量, 如果此变量的值非负，那么分配该资源的权限。如果是负值，那么线程被挂起，直到有其他的线程释放资源；
（2）当线程不再必要该共享资源，必须通过 release() 释放。这样，信号量的内部变量增长，在信号量等候队列中排在最前面的线程会拿到共享资源的权限。

一般说来，为了获得共享资源，线程必要实行下列操作：

测试控制该资源的信号量。
若此信号量的值为正，则允许进行使用该资源。线程将信号量减 1。
若此信号量为 0，则该资源现在不可用，线程进入睡眠状态，直至信号量值大于 0，历程被叫醒，转入步调 1。
当线程不再使用一个信号量控制的资源时，信号量值加 1。如果此时有线程正在睡眠等候此信号量，则叫醒此线程。

在以下的代码中，我们使用生产者-消费者模型展示通过信号量的同步。我们有两个线程， producer() 和 consumer()，它们使用共同的资源 item。 producer() 的任务是生产 item，consumer() 的任务是消费 item。当 item 还没有被生产出来，consumer() 一直等候，当 item 生产出来， producer() 线程关照消费者资源可以使用了。

import threading
import time
import random
_# 通过将信号量初始化为0，我们得到一个所谓的信号量事件_
_# Semaphore可选参数给出内部变量的初始值,默认为1。_
_# 如果给定的值小于0，则抛出ValueError。_
semaphore = threading.Semaphore(0)
def consumer():
print("consumer is waiting.")
if semaphore.acquire():
_# 如果消费者获取到信号量，即信号量大于0_
print("Consumer notify : consumed item number %s " % item)
_# 如果消费者没有获取到信号量，即信号量等于0，此时消费者线程阻塞_
def producer():
global item
time.sleep(1)
item = random.randint(0, 1000)
print("producer notify : produced item number %s" % item)
_# 释放信号量，通知消费者线程_
semaphore.release()
if __name__ == '__main__':
for i in range (0,5):
t1 = threading.Thread(target=producer)
t2 = threading.Thread(target=consumer)
t1.start()
t2.start()
t1.join()
t2.join()
print("program terminated")

复制代码

看一下代码运行结果：

信号量的一个特殊用法是互斥量。互斥量是初始值为 1 的信号量，可以实现数据、资源的互斥访问。信号量在支持多线程的编程语言中依然应用很广，然而这可能导致死锁的环境。
threading.BoundedSemaphore 用于实现有界信号量。有界信号量通过查抄以确保它当前的值不会凌驾初始值。如果凌驾了初始值，将会引发 ValueError 异常。在大多环境下，信号量用于保护数量有限的资源。如果信号量被释放的次数过多，则表明出现了错误。没有指定时， value 的值默以为 1。

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。

		自动登录	找回密码
密码			立即注册

一文速通Python并行计算：03 Python多线程编程-多线程同步（上）—基于互斥 ...

本帖子中包含更多资源

0 个回复

快速回复

楼主热帖

标签云

浏览过的版块

一文速通Python并行计算：03 Python多线程编程-多线程同步（上）—基于互斥 ...

本帖子中包含更多资源

0 个回复

快速回复

楼主热帖

标签云

浏览过的版块

登录参与点评抽奖加入IT实名职场社区