Python多线程与同步机制浅析

更新时间：2022年12月22日 16:38:42 作者：alwaysrun

线程（Thread）是操作系统能够进行运算调度的最小单位；线程自己不拥有系统资源，只拥有一点儿在运行中必不可少的资源，但它可与同属一个进程的其它线程共享进程所拥有的全部资源

线程实现

Python中线程有两种方式：函数或者用类来包装线程对象。threading模块中包含了丰富的多线程支持功能：

threading.currentThread(): 返回当前线程；
threading.enumerate(): 返回包含正在运行的线程列表；
threading.activeCount(): 返回正在运行的线程数量，与len(threading.enumerate())等价。

Thread类

通过Thread类来处理线程，类中提供的一些方法：

run(): 用以表示线程执行的方法（可重载实现实际功能）；
start(): 启动线程；
join([time]): 等待线程中止（或者超时）；
isAlive(): 返回线程是否活动；
getName(): 返回线程名；
setName(): 设置线程名；
setDaemon(True)：设置为后台进程（必须在start调用前设定）。

函数方式

通过Thread直接构造线程，然后通过start方法启动线程：

threading.Thread(group=None, target=None, name=None, args=(), kwargs=None, *,daemon=None)

各参数说明：

group：指定线程隶属的线程组（当前忽略）；
target：指定线程要调度的目标方法（即实现功能的函数）；
args：传递给目标方法的参数（以元组的方式）；
kwargs：传递给目标方法的参数（以字典的方式）；
daemon：指定线程是否为后台线程。

def simpleRoutine(name, delay):
    print(f"routine {name} starting...")
    time.sleep(delay)
    print(f"routine {name} finished")
if __name__ == '__main__':
    thrOne = threading.Thread(target=simpleRoutine, args=("First", 1))
    thrTwo = threading.Thread(target=simpleRoutine, args=("Two", 2))
    thrOne.start()
    thrTwo.start()
    thrOne.join()
    thrTwo.join()

继承方式

直接继承Thread，创建一个新的子类（主要实现run方法）：

class SimpleThread (threading.Thread):
    def __init__(self, name, delay):
        # threading.Thread.__init__(self)
        super().__init__()
        self.name = name
        self.delay = delay
    def run(self):
        print(f"thread {self.name} starting...")
        time.sleep(self.delay)
        print(f"thread {self.name} finished")
if __name__ == '__main__':
    thrOne = SimpleThread("First", 2)
    thrTwo = SimpleThread("Second", 2)
    thrOne.start()
    thrTwo.start()
    thrOne.join()
    thrTwo.join()

同步机制

当多个线程同时修改同一条数据时可能会出现脏数据；所以，就需要线程锁，即同一时刻只允许一个线程执行操作。

同步锁Lock

threading提供了Lock和RLock（可重入锁）两个类，它们都提供了如下两个方法来加锁和释放锁：

acquire(blocking=True, timeout=-1)：加锁，其中 timeout 参数指定加锁多少秒。
release()：释放锁。

两种使用锁的方式：

gCount = 0
def PlusOne(locker):
    global gCount
      with locker:
          gCount += 1、
def MinusOne(locker):
    global gCount
      if locker.acquire():
          gCount -= 1
          locker.release()

条件变量Condition

Condition对象内部维护了一个锁（构造时可传递一个Lock/RLock对象，否则内部会自行创建一个RLock)和一个waiting池：

通过acquire获得Condition对象；
当调用wait方法时，线程会释放Condition内部的锁并进入blocked状态，同时在waiting池中记录这个线程；
当调用notify方法时，Condition对象会从waiting池中挑选一个线程，通知其调用acquire方法尝试取到锁。

Condition对象：

__init__(self,lock=None)：Condition类总是与一个锁相关联（若不指定lock参数，会自动创建一个与之绑定的RLock对象）；

acquire(timeout)：调用关联锁的acquire()方法；

release()：调用关联锁的release()方法

wait(timeout)：线程挂起，直到收到一个notify通知或超时才会被唤醒；必须在已获得锁的前提下调用；

notify(n=1)：唤醒waiting池中的n个正在等待的线程并通知它：

收到通知的线程将自动调用acquire()方法尝试加锁；
若waiting池中有多个线程，随机选择n个唤醒；
必须在已获得锁的前提下调用，否则将引发错误。

notify_all()：通知所有线程。

class Producer(threading.Thread):
    def __init__(self, cond, storage):
        threading.Thread.__init__(self)
        self.cond = cond
        self.storage = storage
    def run(self):
        label = 1
        while True:
            with self.cond:
                if len(self.storage) < 10:
                    self.storage.append(label)
                    print(f"<- Produce {label} product")
                    label += 1
                    self.cond.notify(2)
                else:
                    print(f"<- storage full: Has Produced {label - 1} product")
                    self.cond.notify_all()
                    self.cond.wait()
                time.sleep(0.4)
class Consumer(threading.Thread):
    def __init__(self, name, cond, storage):
        threading.Thread.__init__(self)
        self.name = name
        self.cond = cond
        self.storage = storage
    def run(self):
        while True:
            if self.cond.acquire():
                if len(self.storage) > 1:
                    pro = self.storage.pop(0)
                    print(f"-> {self.name} consumed {pro}")
                    self.cond.notify()
                else:
                    print(f"-> {self.name} storage empty: no product to consume")
                    self.cond.wait()
                self.cond.release()
                time.sleep(1)

信号量Semaphore

信号量对象内部维护一个计数器：

acquire(blocking=True,timeout=None)时减1，当计数为0就阻塞请求的线程；
release()时加1，当计数大于0恢复被阻塞的线程；

threading中有Semaphore和BoundedSemaphore两个信号量；BoundedSemaphore限制了release的次数，任何时候计数器的值，都不不能大于初始值（release时会检测计数器的值，若大于等于初始值，则抛出ValueError异常）。

通过Semaphore维护生产（release一个）、消费（acquire一个）量：

# products = threading.Semaphore(0)
def produceOne(label, sem: threading.Semaphore):
    sem.release()
    print(f"{label} produce one")
def consumeOne(label, sem: threading.Semaphore):
    sem.acquire()
    print(f"{label} consume one")

通过BoundedSemaphore来控制并发数量（最多有Semaphore初始值数量的线程并发）：

# runner = threading.BoundedSemaphore(3)
def runBound(name, sem: threading.BoundedSemaphore):
    with sem:
        print(f"{name} is running")
        time.sleep(1)
        print(f"{name} finished")

事件Event

事件对象内部有个标志字段，用于线程等待事件的发生：

isSet()：返回event的状态值；
wait()：状态为False时，一直阻塞；否则立即返回；
set()：设置状态值为True，激活所有被阻塞的线程；
clear()：恢复状态值为False。

多线程等待事件发生，然后开始执行：

def waiters(name, evt: threading.Event):
    evt.wait()
    print(f"{name} is running")
    time.sleep(1)
    print(f"{name} finished")
def starting(evt: threading.Event):
    evt.set()
    print("event is set")

屏障Barrier

屏障用于设定等待线程数量，当数量达到指定值时，开始执行：

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

屏障属性与方法：

wait(timeout=None)：等待通过屏障；线程被阻塞，直到阻塞的数量达到parties时，被阻塞的线程被同时全部释放；
reset()：重置屏障到默认的空状态；
abort()：将障碍置为断开状态；导致等待的线程引发BrokenBarrierError异常；
partier()：通过障碍所需的线程数；
n_waiting()：当前在屏障中等待的线程数；
broken()：如果屏障处于断开状态，则返回True。

def waitBarrier(name, barr: threading.Barrier):
    print(f"{name} waiting for open")
    try:
        barr.wait()
        print(f"{name} running")
        time.sleep(5)
    except threading.BrokenBarrierError:
        print(f"{name} exception")
    print(f"{name} finished")

GIL全局解释器锁

GIL（Global Interpreter Lock，全局解释器锁）；cpython中，某个线程想要执行，必须先拿到GIL（可以把GIL看作是“通行证”）。每次释放GIL锁，线程都要进行锁竞争，切换线程，会消耗资源。

由于GIL锁的存在，python里一个进程永远只能同时执行一个线程(拿到GIL的线程)，这就是为什么在多核CPU上，python的多线程效率并不高：

CPU密集型代码：由于计算工作多，会很快用完时间片，然后触发GIL的释放与再竞争；
IO密集型代码(文件处理、网络爬虫等)：多线程能够有效提升效率(单线程下有IO操作会进行IO等待，造成不必要的时间浪费，而开启多线程能在线程A等待时，自动切换到线程B，可以不浪费CPU的资源，从而能提升程序执行效率)。

python在使用多线程的时候，调用的是c语言的原生线程：

拿到公共数据
申请GIL
python解释器调用os原生线程
os操作cpu执行运算
当线程执行时间到后，就进行切换（context switch）

到此这篇关于Python多线程与同步机制浅析的文章就介绍到这了,更多相关Python多线程内容请搜索脚本之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持脚本之家！

您可能感兴趣的文章:

如何使用OpenCV实现手势音量控制
今天来学习一下如何使用OpenCV实现手势音量控制,本次实验需要使用OpenCV和mediapipe库进行手势识别,并利用手势距离控制电脑音量,感兴趣的朋友跟随小编一起看看吧
2023-11-11
Python调用C++,通过Pybind11制作Python接口
今天小编就为大家分享一篇关于Python调用C++,通过Pybind11制作Python接口，小编觉得内容挺不错的，现在分享给大家，具有很好的参考价值，需要的朋友一起跟随小编来看看吧
2018-10-10
pytorch 更改预训练模型网络结构的方法
今天小编就为大家分享一篇pytorch 更改预训练模型网络结构的方法，具有很好的参考价值，希望对大家有所帮助。一起跟随小编过来看看吧
2019-08-08
python中的循环结构问题
这篇文章主要介绍了python中的循环结构问题，具有很好的参考价值，希望对大家有所帮助。如有错误或未考虑完全的地方，望不吝赐教
2023-03-03
Python struct模块解析
我们知道python只定义了6种数据类型，字符串，整数，浮点数，列表，元组，字典。但是C语言中有些字节型的变量，在python中该如何实现呢？这点颇为重要，特别是要在网络上进行数据传输的话。
2014-06-06
python 装饰器功能以及函数参数使用介绍
之前学习编程语言大多也就是学的很浅很浅，基本上也是很少涉及到装饰器这些的类似的内容。总是觉得是一样很神奇的东西，舍不得学（嘿嘿）。今天看了一下书籍。发现道理还是很简单的
2012-01-01
python 爬虫网页登陆的简单实现
这篇文章主要介绍了python 爬虫网页登陆的简单实现，文中通过示例代码介绍的非常详细，对大家的学习或者工作具有一定的参考学习价值，需要的朋友们下面随着小编来一起学习学习吧
2020-11-11
进一步探究Python的装饰器的运用
这篇文章主要介绍了更为深入的Python的装饰器的运用,Python的装饰器是Python学习进阶当中的重要知识点,需要的朋友可以参考下
2015-05-05
python 如何对Series中的每一个数据做运算
这篇文章主要介绍了python 实现对Series中的每一个数据做运算操作，具有很好的参考价值，希望对大家有所帮助。如有错误或未考虑完全的地方，望不吝赐教
2021-05-05
django 多数据库配置教程
今天小编就为大家分享一篇django 多数据库配置教程，具有很好的参考价值，希望对大家有所帮助。一起跟随小编过来看看吧
2018-05-05