一文速通Python并行计算：04 Python多线程编程-多线程同步（下）

一文速通 Python 并行计算：04 Python 多线程编程-多线程同步（下）—基于条件变量、事件和屏障

摘要：

本文介绍了 Python 多线程同步的三种机制：条件变量（Condition）、事件（Event）和屏障（Barrier），条件变量指的是线程等待特定条件满足后执行，适用于生产者-消费者模型；Event 指的是线程通过事件标志进行同步，适用于线程间简单通信；Barrier 指的是多个线程需同步到同一阶段时使用，适用于并行任务的分阶段执行。

关于我们更多介绍可以查看云文档：Freak 嵌入式工作室云文档，或者访问我们的 wiki：****https://github.com/leezisheng/Doc/wik

原文链接：

FreakStudio的博客

全网最适合入门的面向对象编程教程：00 面向对象设计方法导论

全网最适合入门的面向对象编程教程：01 面向对象编程的基本概念

全网最适合入门的面向对象编程教程：02 类和对象的Python实现-使用Python创建类

全网最适合入门的面向对象编程教程：03 类和对象的Python实现-为自定义类添加属性

全网最适合入门的面向对象编程教程：04 类和对象的 Python 实现-为自定义类添加方法

全网最适合入门的面向对象编程教程：05 类和对象的Python实现-PyCharm代码标签

全网最适合入门的面向对象编程教程：06 类和对象的 Python 实现-自定义类的数据封装

全网最适合入门的面向对象编程教程：07 类和对象的Python实现-类型注解

全网最适合入门的面向对象编程教程：08 类和对象的 Python 实现-@property 装饰器

全网最适合入门的面向对象编程教程：09 类和对象的Python实现-类之间的关系

全网最适合入门的面向对象编程教程：10 类和对象的Python实现-类的继承和里氏替换原则

全网最适合入门的面向对象编程教程：11 类和对象的Python实现-子类调用父类方法

全网最适合入门的面向对象编程教程：12 类和对象的 Python 实现-Python 使用 logging 模块输出程序运行日志

全网最适合入门的面向对象编程教程：13 类和对象的 Python 实现-可视化阅读代码神器 Sourcetrail 的安装使用

全网最适合入门的面向对象编程教程：14 类和对象的Python实现-类的静态方法和类方法

全网最适合入门的面向对象编程教程：15 类和对象的Python实现-__slots__魔法方法

全网最适合入门的面向对象编程教程：16 类和对象的Python实现-多态、方法重写与开闭原则

全网最适合入门的面向对象编程教程：17 类和对象的Python实现-鸭子类型与“file-like object“

全网最适合入门的面向对象编程教程：18 类和对象的Python实现-多重继承与PyQtGraph串口数据绘制曲线图

全网最适合入门的面向对象编程教程：19 类和对象的 Python 实现-使用 PyCharm 自动生成文件注释和函数注释

全网最适合入门的面向对象编程教程：20 类和对象的 Python 实现-组合关系的实现与 CSV 文件保存

全网最适合入门的面向对象编程教程：21 类和对象的 Python 实现-多文件的组织：模块 module 和包 package

全网最适合入门的面向对象编程教程：22 异常捕获-异常和语法错误

全网最适合入门的面向对象编程教程：23 异常捕获-抛出异常

全网最适合入门的面向对象编程教程：24 异常捕获现-异常的捕获与处理：try/except语句、文件读写示例、Exception引用

全网最适合入门的面向对象编程教程：25 异常捕获-Python 判断输入数据类型

全网最适合入门的面向对象编程教程：26 异常捕获-上下文管理器和with语句

全网最适合入门的面向对象编程教程：27 异常捕获-Python 中异常层级与自定义异常类的实现

全网最适合入门的面向对象编程教程：28 异常捕获-Python编程原则、哲学和规范大汇总

全网最适合入门的面向对象编程教程：29 异常捕获-断言与防御性编程和help函数的使用

全网最适合入门的面向对象编程教程：30 Python的内置数据类型-object根类

全网最适合入门的面向对象编程教程：31 Python的内置数据类型-对象Object和类型Type

全网最适合入门的面向对象编程教程：32 Python的内置数据类型-类Class和实例Instance

全网最适合入门的面向对象编程教程：33 Python 的内置数据类型-对象 Object 和类型 Type 的关系

全网最适合入门的面向对象编程教程：34 Python的内置数据类型-Python常用复合数据类型：元组和命名元组

全网最适合入门的面向对象编程教程：35 Python的内置数据类型-文档字符串和__doc__属性

全网最适合入门的面向对象编程教程：36 Python的内置数据类型-字典

全网最适合入门的面向对象编程教程：37 Python常用复合数据类型-列表和列表推导式

全网最适合入门的面向对象编程教程：38 Python常用复合数据类型-使用列表实现堆栈、队列和双端队列

全网最适合入门的面向对象编程教程：39 Python 常用复合数据类型-集合

全网最适合入门的面向对象编程教程：40 Python 常用复合数据类型-枚举和 enum 模块的使用

全网最适合入门的面向对象编程教程：41 Python常用复合数据类型-队列（FIFO、LIFO、优先级队列、双端队列和环形队列）

全网最适合入门的面向对象编程教程：42 Python常用复合数据类型-collections容器数据类型

全网最适合入门的面向对象编程教程：43 Python常用复合数据类型-扩展内置数据类型

全网最适合入门的面向对象编程教程：44 Python内置函数与魔法方法-重写内置类型的魔法方法

全网最适合入门的面向对象编程教程：45 Python实现常见数据结构-链表、树、哈希表、图和堆

全网最适合入门的面向对象编程教程：46 Python函数方法与接口-函数与事件驱动框架

全网最适合入门的面向对象编程教程：47 Python函数方法与接口-回调函数Callback

全网最适合入门的面向对象编程教程：48 Python函数方法与接口-位置参数、默认参数、可变参数和关键字参数

全网最适合入门的面向对象编程教程：49 Python函数方法与接口-函数与方法的区别和lamda匿名函数

全网最适合入门的面向对象编程教程：50 Python函数方法与接口-接口和抽象基类

全网最适合入门的面向对象编程教程：51 Python函数方法与接口-使用Zope实现接口

全网最适合入门的面向对象编程教程：52 Python函数方法与接口-Protocol协议与接口

全网最适合入门的面向对象编程教程：53 Python字符串与序列化-字符串与字符编码

全网最适合入门的面向对象编程教程：54 Python字符串与序列化-字符串格式化与format方法

全网最适合入门的面向对象编程教程：55 Python字符串与序列化-字节序列类型和可变字节字符串

全网最适合入门的面向对象编程教程：56 Python字符串与序列化-正则表达式和re模块应用

全网最适合入门的面向对象编程教程：57 Python字符串与序列化-序列化与反序列化

全网最适合入门的面向对象编程教程：58 Python字符串与序列化-序列化Web对象的定义与实现

全网最适合入门的面向对象编程教程：59 Python并行与并发-并行与并发和线程与进程

一文速通Python并行计算：00 并行计算的基本概念

一文速通Python并行计算：01 Python多线程编程-基本概念、切换流程、GIL锁机制和生产者与消费者模型

一文速通Python并行计算：02 Python多线程编程-threading模块、线程的创建和查询与守护线程

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

文档获取：

可访问如下链接进行对文档下载：

https://github.com/leezisheng/Doc

该文档是一份关于 并行计算 和 Python 并发编程 的学习指南，内容涵盖了并行计算的基本概念、Python 多线程编程、多进程编程以及协程编程的核心知识点：

正文

1.基于条件变量的线程同步

Python 提供的 Condition 对象提供了对复杂线程同步问题的支持。Condition 被称为条件变量，与互斥锁不同，条件变量是用来等待而不是用来上锁的。**条件变量用来自动阻塞一个线程，直到某特殊情况发生为止。**条件变量是利用线程间共享的全局变量进行同步的一种机制，主要包括两个动作：

一个线程等待"条件变量的条件成立"而挂起；
另一个线程使 “条件成立”（给出条件成立信号）。

使用 Condition 的主要方式为：**线程首先acquire一个条件变量，然后判断一些条件。如果条件不满足则wait；如果条件满足，进行一些处理改变条件后，通过notify方法通知其他线程，其他处于wait状态的线程接到通知后会重新判断条件，**不断的重复这一过程，从而解决复杂的同步问题。

解释条件机制最好的例子还是生产者-消费者问题，在本例中，只要缓存不满，生产者一直向缓存生产；只要缓存不空，消费者一直从缓存取出（之后销毁）；当缓冲队列不为空的时候，生产者将通知消费者；当缓冲队列不满的时候，消费者将通知生产者。

**i**mport threading
import time

condition = threading.Condition()
products = 0

class Producer(threading.Thread):
    def __init__(self):
        threading.Thread.__init__(self)

    def run(self):
        global condition, products
        while True:
            _# 消费者通过拿到锁来修改共享的资源_
            if condition.acquire(): 
            _# 线程首先acquire一个条件变量，然后判断生产是否饱和。_
                if products < 10:
                    _# 如果产品数量小于10，继续生成，并通过notify方法通知消费者_
                    _# 只要缓存不满，生产者一直向缓存生产；_
                    products += 1;
                    print("Producer(%s):deliver one, now products:%s" % (self.name, products))
                    _# 当缓冲队列不为空的时候，生产者将通知消费者_
                    condition.notify()
                _# 如果已经满了，那么生产者进入等待状态，直到被唤醒_
                else:
                    print("Producer(%s):already 10, stop deliver, now products:%s" % (self.name, products))
                    condition.wait()
                _# 如果队列没有满，就生产1个item，通知状态并释放资源_
                condition.release()
                time.sleep(2)

class Consumer(threading.Thread):
    def __init__(self):
        threading.Thread.__init__(self)

    def run(self):
        global condition, products
        while True:
            if condition.acquire():
                if products > 1:
                    _# 只要缓存不空，消费者一直从缓存取出（之后销毁）。_
                    products -= 1
                    print("Consumer(%s):consume one, now products:%s" % (self.name, products))
                    _# 当缓冲队列不满的时候，消费者将通知生产者。_
                    condition.notify()
                else:
                    print("Consumer(%s):only 1, stop consume, products:%s" % (self.name, products))
                    _# 缓存空，消费者线程等待_
                    condition.wait()
                condition.release()
                time.sleep(2)

if __name__ == "__main__":
    _# 首先是2个生成者生产products_
    for p in range(0, 2):
        p = Producer()
        p.start()
    _# 接下来的10个消费者将会消耗products_
    for c in range(0, 10):
        c = Consumer()
        c.start()

运行的结果如下：

乍一看这段代码好像会死锁，因为 condition.acquire() 之后就在 wait() 了，好像会一直持有锁。其实 wait() 会将锁释放，然后等待其他线程 notify() 之后会重新尝试获得锁。**但是要注意 notify() 并不会自动释放锁，所以代码中有两行，先 notify() 然后再 ****release() **。

实际上，**条件的检测是在互斥锁的保护下进行的。线程在改变条件状态之前必须首先锁住互斥量。**如果一个条件为假，一个线程自动阻塞，并释放等待状态改变的互斥锁。**如果另一个线程改变了条件，它发信号给关联的条件变量，唤醒一个或多个等待它的线程，重新获得互斥锁，重新评价条件。**如果两进程共享可读写的内存，条件变量可以被用来实现这两进程间的线程同步。

另外：Condition对象的构造函数可以接受一个Lock/RLock对象作为参数，如果没有指定，则Condition对象会在内部自行创建一个RLock**；**除了 notify 方法外，Condition 对象还提供了 notifyAll 方法，可以通知 waiting 池中的所有线程尝试 acquire 内部锁。由于上述机制，处于 waiting 状态的线程只能通过 notify 方法唤醒，所以 notifyAll 的作用在于防止有线程永远处于沉默状态。

**如果不使用条件变量，也可以不断循环检测缓存是否大于 0，但该方法会造成 CPU 资源的浪费。**采用条件变量这一问题就可以迎刃而解！条件变量允许一个线程休眠（阻塞等待）直至获取到另一个线程的通知（收到信号）再去执行自己的操作。

以上消费者-生产者模型过程如下：

以下是另一个有趣的关于条件变量的例子：

原理很简单，就是线程拿到锁先检查是不是自己渴望的状态。比如打印“B”的线程，渴望的状态 current = 'B' 然后打印出 B，将状态改成 C ，这样就成了打印“C”的线程渴望的状态。但是这里不能唤醒指定的线程，只好唤醒所有的线程，让他们自己再检查一遍状态了。

2.基于事件的线程同步

想象这样一个场景，你启动了多个线程，这些线程都要去访问一个资源，但是，这里有一个小小的问题，即将被访问的资源还没有准备好接受访问，那么此时，多个线程去访问，必然得到不响应，你还得处理这种得不到响应的情况。这样的场景下，能否先在主线程里去做试探，确定资源可以访问以后，再让已经启动了的多线程去访问呢？让我们考虑一下如何用 Event 来处理这样的问题。

创建一个 Event 对象，现在，事件内部标识是 False；
启动多线程，线程里调用 wait 方法，这时，会阻塞；
主线程去试探，确定资源可访问以后，调用 set 方法，将内置标志设置为 True；
该 Event 会通知所有等待状态的线程恢复运行：已经调用 wait 的线程接手到事件信息，访问资源。

以下为示例代码：

import threading
from threading import Event


def worker(event_obj, i):
    print('{i}号线程等待事件信号'.format(i=i))
    event_obj.wait()
    print('{i}号线程收到事件信号'.format(i=i))


event = Event()

for i in range(5):
    t = threading.Thread(target=worker, args=(event, i))
    t.start()

print('确认资源可用')
event.set()

以下为代码输出，可以看到在 event.set() 后所有线程恢复运行：

让我们再一次回到生产者-消费者问题上，若要确保如果缓冲区满，生产者不会生成新数据，另外如果缓存区为空，消费者不会查找数据的要求，如何用 event 机制实现？代码如下：

import time
from threading import Thread, Event
import random

items = []
event = Event()

class consumer(Thread):
    def __init__(self, items, event):
        _#  producer 类初始化时定义了item的list和 Event ，_
        _# 与条件对象时候的例子不同，这里的list并不是全局的，而是通过参数传入的_
        Thread.__init__(self)
        self.items = items
        self.event = event

    def run(self):
        while True:
            time.sleep(2)
            _# 等待元素到达_
            _# 当元素到达时，consumer放弃这个锁_
            _# 这就允许其他生产者或消费者进入并获得这个锁_
            self.event.wait()

            _# 若consumer放弃这个锁被唤醒，它会重新获得锁_
            _# 元素到达时，元素从items列表弹出_
            item = self.items.pop()
            print('Consumer notify : %d popped from list by %s' % (item, self.name))

class producer(Thread):
    def __init__(self, items, event):
        Thread.__init__(self)
        self.items = items
        self.event = event

    def run(self):
        global item
        for i in range(100):
            time.sleep(2)
            item = random.randint(0, 256)
            self.items.append(item)
            print('Producer notify : item N° %d appended to list by %s' % (item, self.name))
            print('Producer notify : event set by %s' % self.name)
            _# 添加元素后通知事件_
            _# 将内部标识设置为 true 。所有正在等待这个事件的线程将被唤醒。_
            _# 当标识为 true 时，调用wait()方法的线程不会被阻塞。_
            self.event.set()
            print('Produce notify : event cleared by %s '% self.name)
            _# 将内部标识设置为 false 。之后调用wait()方法的线程将会被阻塞，_
            _# 直到调用set()方法将内部标识再次设置为 true 。_
            self.event.clear()

if __name__ == '__main__':
    t1 = producer(items, event)
    t2 = consumer(items, event)
    t1.start()
    t2.start()
    t1.join()
    t2.join()

上图是运行程序时候的运行结果，线程 t1 在 list 最后添加值，然后设置 event 来通知消费者。消费者通过 wait() 阻塞，直到收到信号的时候从 list 中取出元素消费。

补充一下 wait() 方法：

3.基于屏障的线程同步

屏障用于应对固定数量的线程需要彼此相互等待的情况，与之前介绍 互斥锁Lock/事件Event/定时器Timer 等不同，多线程Barrier会设置一个线程障碍数量parties，如果等待的线程数量没有达到障碍数量 parties，所有线程会处于阻塞状态，当等待的线程到达了这个数量就会唤醒所有的等待线程。

以播放器为例子：

首先一个线程做播放器初始化工作(加载本地文件或者获取播放地址)，然后一个线程获取视频画面，一个线程获取视频声音，只有当初始化工作完毕，视频画面获取完毕，视频声音获取完毕，播放器才会开始播放，其中任意一个线程没有完成，播放器会处于阻塞状态直到三个任务都完成！

class threading.Barrier(parties, action=None, timeout=None)

创建一个需要 parties 个线程的栅栏对象。如果提供了可调用的 action 参数，它会在所有线程被释放时在其中一个线程中自动调用。 timeout 是默认的超时时间，如果没有在 wait() 方法中指定超时时间的话。

使用方法包括：

以下为示例代码，创建三个线程：初始化准备、音频准备、视频准备，当且仅当三个初始化完成，才能启动音乐播放。

_# 导入线程模块_
import threading

def plyer_display():
    print('初始化通过完成，音视频同步完成，可以开始播放....')

_# 设置3个障碍对象_
barrier = threading.Barrier(3, action=plyer_display, timeout=None)
def player_init(statu):
    print(statu)
    try:
        _# 设置超时时间，如果2秒内，没有达到障碍线程数量，_
        _# 会进入断开状态，引发BrokenBarrierError错误_
        barrier.wait(2)
    except Exception as e:
    _# 断开状态，引发BrokenBarrierError错误_
            print("等待超时了... ")
    else:
        print("xxxooooxxxxxooooxxxoooo")
        
if __name__ == '__main__':
    statu_list = ["init ready", "video ready", "audio ready"]
    thread_list = list()
    _# 创建三个线程：初始化准备、音频准备、视频准备_
    _# 当且仅当三个初始化完成，才能启动音乐播放_
    for i in range(0, 3):
        t = threading.Thread(target=player_init, args=(statu_list[i],))
        t.start()
        thread_list.append(t)
    for t in thread_list:
        t.join()

以下为代码运行结果，注意：如果 barrier.wait(timeout=None) 等待超时，会进入断开状态，引发 BrokenBarrierError 错误，为了程序的健壮性，最好加上异常处理。

一文速通Python并行计算：04 Python多线程编程-多线程同步（下）—基于条件变量、事件和屏障