当前位置：首页>后端>正文

Python 异步 IO 并发编程详解

后端2024-04-20 07:41:08

一、多线程、多进程和线程池编程
- 1.1 关于 Python 中的 GIL（全局解释器锁）
- 1.2 Python 多线程编程 threading
- 1.3 线程间通信：共享变量和 Queue
- 1.4 线程同步：Lock、RLock
- 1.5 线程同步：condition、Semaphore 使用
- 1.6 ThreadPoolExecutor 线程池及相关分析
- 1.7 multiprocessing 多进程编程及与多线程对比
二、协程和异步 IO
- 2.1 并发、并行、同步、异步、阻塞、非阻塞
- 2.2 什么是协程
- 2.3 生成器进阶：send、close 和 throw 方法
- 2.4 生成器进阶：yield from
三、asyncio 并发编程
- 3.1 asyncio 介绍
- 3.2 future 和 task 讲解
- 3.3 asyncio 函数回调
- 3.4 asyncio 多任务执行
- 3.4 asyncio call_soon 方法
- 3.5 asyncio 锁
结束语
大家好，本次写作主要给大家讲解 Python 的异步 IO 并发编程三个部分。分别是多线程、多进程和线程池编程、协程和异步 IO、asyncio 并发编程。希望大家通过本文能对 Python 的异步 IO 并发编程有一个全新的认识。

本文读者群体：具有一定 Python 基础的读者。

一、多线程、多进程和线程池编程

1.1 关于 Python 中的 GIL（全局解释器锁）

GIL 全局解释器锁是我们学习 Python 的多进程和多线程知识点第一个需要了解的知识点，因为本次 Python 讲解是基于 CPython 解释器，而 Python 的一个线程对应着 C 语言中的一个线程。因为多个线程运行同一段代码可能会导致运行出错，GIL 的存在就是使同一个时刻只有一个线程在一个 CPU 上执行字节码，因此多线程编程时无法将多个线程映射到多个 CPU 上去执行，使得 CPU 资源无法完全被利用。但是 GIL 也不是完全不释放，GIL 的释放主要在以下两个方面。

1. GIL 会根据执行的字节码和时间片段释放 GIL，例如：

from threading import Thread
total=0
def add():
    global total;
    for i in range(1000000):
        total +=1

def desc():
    global total
    for i in range(1000000):

        total-=1

t1 = Thread(target=add)
t2 = Thread(target=desc)

t1.start()
t2.start()

t1.join()
t2.join()

print(total)

上述结果每次会因为 GIL 释放而得到的结果不同。

2. GIL 在遇到 IO 操作的时候会主动释放，这个使得 Python 在处理相关 IO 方面极具优势。

1.2 Python 多线程编程 threading

thread 模块在 Python 中属于比较底层的模块，而 threading 实际上是对 thread 做了一些包装，从而可以更加方便被使用。

简单示例 1：

from threading import Thread
import time
def study():
    print("hello threading")
    time.sleep(2)

if __name__ == "__main__":

    for i in range(5):
        t = Thread(target=study)
        #启动线程
        t.start()

简单示例 2:

from threading import Thread
import time

class Study(Thread):

    def run(self):

        print("study threading....")
        msg = "I am "+self.name+"@"+str(i)
        time.sleep(2)
        print(msg)

if __name__=="__main__":
    for i in range(5):
        t1 = Study()
        t1.start()

通过上面这个示例可以看出使用多线程并发的操作，花费时间要比单线程花费的时间小的多，其实对于 IO 来说线程和进程的性能差别不大。

注意点：Python 中主线程会等待所有的子线程结束后才结束。

1.3 线程间通信：共享变量和 Queue

共享变量：进程内所有的多线程对于全局变量进行共享。

优点：所有线程共享全局变量，便于数据传递
缺点：由于全局变量能随意修改容易造成数据混乱，导致一些未知的错误

关于举例其实在全局 GIL 中大家也可以看出，这里我再重新给大家举一个例子。

示例：

from threading import Thread
from time import sleep

nums = 100

def setNums():
    global nums
    for i in range(5):
        nums +=1

    print("in setNums nums is %d"%nums)

def getNums():
    global nums

    print("in getNums nums is %d"%nums)

print("线程创建之前:%d"%nums)

t1 = Thread(target=setNums)
t1.start()

sleep(3)

t2 = Thread(target=getNums)
t2.start()

运行结果：

    线程创建之前:100
    in setNums nums is 105
    in getNums nums is 105

这里可以看到两个线程获取的 nums 值都是 105。

Queue：Python 的 Queue 模块中提供了以下几种队列类：

FIFO（先入先出) 队列 Queue
LIFO（后入先出）队列 LifoQueue
优先级队列 Priority Queue

一般我们可以使用队列来实现线程同步，在开发中 FIFO 队列我们使用的比较多，下面我将用一个例子说明：

from threading import Thread
from time import sleep
from queue import Queue

#生产者
def Producer():
    count =0
    while True:
        if queue.qsize()<1000:
            for i in range(100):
                count +=1
                msg = "生产商品"+str(count)
                queue.put(msg)
                print(msg)

        sleep(0.5)

#消费者
def Consumer():
    while True:
        if queue.qsize()>100:
            for i in range(3):
                msg = "消费者消费了"+queue.get()
                print(msg)

        sleep(1)

if __name__=="__main__":
    #定义一个队列
    queue = Queue();

#初始化商品
for i in range(500):
    queue.put("初始商品"+str(i))
   #生产商品
    for i in range(4):
        p = Thread(target=Producer)
        p.start()
    #消费商品
    for i in range(10):
        c = Thread(target=Consumer)
        c.start()

队列对象（Queue、LifoQueue 或者 PriorityQueue）提供下列描述的公共方法。

Queue.qsize()

返回队列的大致大小。注意，qsize()> 0 不保证后续的 get() 不被阻塞，qsize() < maxsize 也不保证 put() 不被阻塞。

Queue.empty()

如果队列为空，返回 True，否则返回 False。如果 empty() 返回 True，不保证后续调用的 put() 不被阻塞。类似的，如果 empty() 返回 False，也不保证后续调用的 get() 不被阻塞。

Queue.full()

如果队列是满的返回 True，否则返回 False。如果 full() 返回 True 不保证后续调用的 get() 不被阻塞。类似的，如果 full() 返回 False 也不保证后续调用的 put() 不被阻塞。

Queue.put(item, block=True, timeout=None)

将 item 放入队列。如果可选参数 block 是 true 并且 timeout 是 None（默认），则在必要时阻塞至有空闲插槽可用。如果 timeout 是个正数，将最多阻塞 timeout 秒，如果在这段时间没有可用的空闲插槽，将引发 Full 异常。反之（block 是 false），如果空闲插槽立即可用，则把 item 放入队列，否则引发 Full 异常（在这种情况下，timeout 将被忽略）。

Queue.put_nowait (item)

相当于 put(item, False)。

Queue.get(block=True, timeout=None)

从队列中移除并返回一个项目。如果可选参数 block 是 true 并且 timeout 是 None（默认值），则在必要时阻塞至项目可得到。如果 timeout 是个正数，将最多阻塞 timeout 秒，如果在这段时间内项目不能得到，将引发 Empty 异常。反之（block 是 false），如果一个项目立即可得到，则返回一个项目，否则引发 Empty 异常（这种情况下，timeout 将被忽略）。

POSIX 系统 3.0 之前，以及所有版本的 Windows 系统中，如果 block 是 true 并且 timeout 是 None，这个操作将进入基础锁的不间断等待。这意味着，没有异常能发生，尤其是 SIGINT 将不会触发 KeyboardInterrupt 异常。

Queue.get_nowait()

相当于 get(False)。提供了两个方法，用于支持跟踪排队的任务是否被守护的消费者线程完整的处理。

Queue.task_done()

查看全文

https://www.xamrdz.com/backend/3yv1923414.html

一、多线程、多进程和线程池编程

1.1 关于 Python 中的 GIL（全局解释器锁）

1.2 Python 多线程编程 threading

1.3 线程间通信：共享变量和 Queue

相关文章：