Python多线程与多进程怎么选？详解GIL锁与多核CPU利用
作者：一只楚楚猫
进程(Process)和线程(Thread)
现代操作系统(Mac OS X，UNIX，Linux，Windows)都是支持“多任务”的操作系统
“多任务”：操作系统可以同时运行多个任务 ⇒ \Rightarrow ⇒打个比方，你一边在用浏览器上网，一边在听MP3，一边在用Word赶作业，至少同时有3个任务正在运行
单核CPU执行多任务的方式：操作系统轮流让各个任务交替执行 → \to →任务1执行0.01秒，切换到任务2，任务2执行0.01秒，再切换到任务3，执行0.01秒……这样反复执行下去
只能在多核CPU上实现「真正的并行执行多任务」
对于操作系统而言，一个任务就是一个进程(Process) ⇒ \Rightarrow ⇒比如打开一个浏览器就是启动一个浏览器进程，打开一个记事本就是启动一个记事本进程
有些进程还不止同时干一件事，比如Word，它可以同时进行打字、拼写检查、打印等事情。在一个进程内部，要同时干多件事，就需要同时运行多个“子任务”，进程内的这些“子任务”被称为线程(Thread)
由于每个进程至少要干一件事，所以，一个进程至少有一个线程
线程是最小的执行单元，而进程由至少一个线程组成
在python中，多任务的实现方式：多进程模式；多线程模式；多进程+多线程模式
多进程
fork()：Unix/Linux操作系统中的系统调用，不同于普通函数的一次调用返回一次，fork()的一次调用将返回两次 ⇒ \Rightarrow ⇒原因：操作系统自动把当前父进程复制一份为子进程，然后分别在父进程和子进程内返回 → \to →子进程永远返回0，而父进程返回子进程的ID
在python的父进程中创建子进程
import os
print('Process (%s) start...' % os.getpid())
# Only works on Unix/Linux/macOS:
pid = os.fork()
if pid == 0:
print('I am child process (%s) and my parent is %s.' % (os.getpid(), os.getppid()))
else:
print('I (%s) just created a child process (%s).' % (os.getpid(), pid))
运行结果
Process (876) start... I (876) just created a child process (877). I am child process (877) and my parent is 876.
在Unix/Linux操作系统中，通过fork()调用，一个进程在接到新任务时就可以复制出一个子进程来处理新任务
multiprocessing
multiprocessing模块通过提供Process类来表示一个进程对象
from multiprocessing import Process
import os
# 子进程要执行的代码
def run_proc(name):
print('Run child process %s (%s)...' % (name, os.getpid()))
if __name__=='__main__':
print('Parent process %s.' % os.getpid())
p = Process(target=run_proc, args=('test',))
print('Child process will start.')
p.start()
p.join()
print('Child process end.')
执行结果
Parent process 928. Child process will start. Run child process test (929)... Process end.
start()：启动进程join()：等待子进程结束后再继续往下运行
Pool
可以通过进程池的方式启动大量的子进程
from multiprocessing import Pool
import os, time, random
def long_time_task(name):
print('Run task %s (%s)...' % (name, os.getpid()))
start = time.time()
time.sleep(random.random() * 3)
end = time.time()
print('Task %s runs %0.2f seconds.' % (name, (end - start)))
if __name__=='__main__':
print('Parent process %s.' % os.getpid())
p = Pool(4)
for i in range(5):
p.apply_async(long_time_task, args=(i,))
print('Waiting for all subprocesses done...')
p.close()
p.join()
print('All subprocesses done.')
执行结果
Parent process 669. Waiting for all subprocesses done... Run task 0 (671)... Run task 1 (672)... Run task 2 (673)... Run task 3 (674)... Task 2 runs 0.14 seconds. Run task 4 (673)... Task 1 runs 0.27 seconds. Task 3 runs 0.86 seconds. Task 0 runs 1.41 seconds. Task 4 runs 1.91 seconds. All subprocesses done.
apply_async()：异步提交任务，任务提交后，主进程可以立即继续执行，不会阻塞等待结果apply()：同步提交任务，必须等函数执行完成才能继续下一步close()：调用之后不能继续添加新的Processjoin()：等待所有子进程执行完毕
子进程
在外部进程作为子进程时，需要控制子进程的输入和输出
subprocess模块：启动一个子进程，然后控制其输入和输出
在python代码中运行命令nslookup www.python.org，这和命令行直接运行的效果是一样的
import subprocess
print('$ nslookup www.python.org')
r = subprocess.call(['nslookup', 'www.python.org'])
print('Exit code:', r)
nslookup(name server lookup)：用于查询 DNS（域名系统） 信息
运行结果
$ nslookup www.python.org Server: 192.168.19.4 Address: 192.168.19.4#53 Non-authoritative answer: www.python.org canonical name = python.map.fastly.net. Name: python.map.fastly.net Address: 199.27.79.223 Exit code: 0
如果子进程还需要输入，则可以通过communicate()方法输入
import subprocess
print('$ nslookup')
p = subprocess.Popen(['nslookup'], stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
output, err = p.communicate(b'set q=mx\npython.org\nexit\n')
print(output.decode('utf-8'))
print('Exit code:', p.returncode)
communicate()：向子进程stdin写入字节串 → \to →等待子进程执行完成 → \to →返回输出和错误信息
上面的代码相当于在命令行执行命令nslookup，然后手动输入
set q=mx # 设置查询类型为 MX（邮件服务器） python.org # 查询 python.org 域名 exit # 退出 nslookup
运行结果
$ nslookup Server: 192.168.19.4 Address: 192.168.19.4#53 Non-authoritative answer: python.org mail exchanger = 50 mail.python.org. Authoritative answers can be found from: mail.python.org internet address = 82.94.164.166 mail.python.org has AAAA address 2001:888:2000:d::a6 Exit code: 0
进程间通信
multiprocessing模块提供了Queue、Pipes等多种方式来交换数据
以Queue为例，在父进程中创建两个子进程，一个往Queue里写数据，一个从Queue里读数据
from multiprocessing import Process, Queue
import os, time, random
# 写数据进程执行的代码:
def write(q):
print('Process to write: %s' % os.getpid())
for value in ['A', 'B', 'C']:
print('Put %s to queue...' % value)
q.put(value)
time.sleep(random.random())
# 读数据进程执行的代码:
def read(q):
print('Process to read: %s' % os.getpid())
while True:
value = q.get(True)
print('Get %s from queue.' % value)
if __name__=='__main__':
# 父进程创建Queue，并传给各个子进程：
q = Queue()
pw = Process(target=write, args=(q,))
pr = Process(target=read, args=(q,))
# 启动子进程pw，写入:
pw.start()
# 启动子进程pr，读取:
pr.start()
# 等待pw结束:
pw.join()
# pr进程里是死循环，无法等待其结束，只能强行终止:
pr.terminate()
运行结果
Process to write: 50563 Put A to queue... Process to read: 50564 Get A from queue. Put B to queue... Get B from queue. Put C to queue... Get C from queue.
多线程
多任务可以由多进程完成，也可以由一个进程内的多线程完成
进程是由若干线程组成的，一个进程至少有一个线程
python的标准库提供了两个模块：_thread和threading，_thread是低级模块，threading是高级模块，对_thread进行了封装
import time, threading
# 新线程执行的代码:
def loop():
print('thread %s is running...' % threading.current_thread().name)
n = 0
while n < 5:
n = n + 1
print('thread %s >>> %s' % (threading.current_thread().name, n))
time.sleep(1)
print('thread %s ended.' % threading.current_thread().name)
print('thread %s is running...' % threading.current_thread().name)
t = threading.Thread(target=loop, name='LoopThread')
t.start()
t.join()
print('thread %s ended.' % threading.current_thread().name)
执行结果
thread MainThread is running... thread LoopThread is running... thread LoopThread >>> 1 thread LoopThread >>> 2 thread LoopThread >>> 3 thread LoopThread >>> 4 thread LoopThread >>> 5 thread LoopThread ended. thread MainThread ended.
由于任何进程默认就会启动一个线程，我们把该线程称为主线程，主线程又可以启动新的线程
Lock
多线程和多进程
- 多进程：同一个变量，各自有一份拷贝存在于每个进程中，互不影响
- 多线程：所有变量都由所有线程共享，任何一个变量都可以被任何一个线程修改
线程共享数据的最大风险在于，当多个线程同时修改同一个变量时，可能会导致数据被破坏或出现错误
# multithread
import time, threading
# 假定这是你的银行存款:
balance = 0
def change_it(n):
# 先存后取，结果应该为0:
global balance
balance = balance + n
balance = balance - n
def run_thread(n):
for i in range(10000000):
change_it(n)
t1 = threading.Thread(target=run_thread, args=(5,))
t2 = threading.Thread(target=run_thread, args=(8,))
t1.start()
t2.start()
t1.join()
t2.join()
print(balance)
我们先定义了一个共享变量 balance，初始值设为 0。然后启动两个线程：一个负责存入，一个负责取出。按理说，最终结果应该仍然是 0。然而，由于线程调度由操作系统控制，当线程 t1 和 t2 交替执行时，如果循环次数足够多，balance 的最终结果就可能偏离 0
高级语言中的一条语句在 CPU 中往往需要分解为多条指令来执行
balance = balance + n
也分两步：
- 计算
balance + n，存入临时变量中 - 将临时变量的值赋给
balance
由于x是局部变量，两个线程各自都有自己的x
初始值 balance = 0
t1: x1 = balance + 5 # x1 = 0 + 5 = 5
t1: balance = x1 # balance = 5
t1: x1 = balance - 5 # x1 = 5 - 5 = 0
t1: balance = x1 # balance = 0
t2: x2 = balance + 8 # x2 = 0 + 8 = 8
t2: balance = x2 # balance = 8
t2: x2 = balance - 8 # x2 = 8 - 8 = 0
t2: balance = x2 # balance = 0
结果 balance = 0
但是t1和t2是交替运行的，如果操作系统以下面的顺序执行t1、t2
初始值 balance = 0 t1: x1 = balance + 5 # x1 = 0 + 5 = 5 t2: x2 = balance + 8 # x2 = 0 + 8 = 8 t2: balance = x2 # balance = 8 t1: balance = x1 # balance = 5 t1: x1 = balance - 5 # x1 = 5 - 5 = 0 t1: balance = x1 # balance = 0 t2: x2 = balance - 8 # x2 = 0 - 8 = -8 t2: balance = x2 # balance = -8 结果 balance = -8
由于修改 balance 并不是一条原子操作，而是由多条语句组成的。在这些语句执行的过程中，线程可能会被切换，从而导致多个线程同时修改同一个对象，结果就把数据弄乱了
为了确保 balance 的计算结果正确，需要在 change_it() 方法外加上一把锁。当某个线程开始执行 change_it() 时，它会先获取锁，因此其他线程必须等待，直到该锁被释放后才能继续执行。
由于同一时间最多只有一个线程能够持有这把锁，就避免了多个线程同时修改 balance 所造成的冲突。在 python 中，可以通过 threading.Lock() 来创建这样一把锁
balance = 0
lock = threading.Lock()
def run_thread(n):
for i in range(100000):
# 先要获取锁:
lock.acquire()
try:
# 放心地改吧:
change_it(n)
finally:
# 改完了一定要释放锁:
lock.release()
当多个线程同时调用 lock.acquire() 时，只有一个线程能成功拿到锁并继续执行，其余线程则会阻塞等待，直到锁被释放后才能依次获取并运行
线程在获取锁后，必须在使用完毕后及时释放。如果不释放，其他等待锁的线程就会一直被阻塞，最终造成死锁。
为避免这种情况，通常会使用 try...finally 结构来确保锁一定能被释放
锁的优点在于，它能保证某段关键代码只能由一个线程完整执行，从而避免并发冲突。但缺点也很明显：首先，锁会阻碍多线程的并行性，使相关代码实际退化为单线程执行，效率因此下降。
其次，如果系统中存在多个锁，不同线程各自持有一部分锁并同时尝试获取对方的锁，就可能陷入死锁状态，导致所有相关线程都无法继续运行，也不能正常结束，只能依靠操作系统强制终止
单例模式
单例模式
- 类只能创建一个实例
- 全局共享这个实例
import threading
class Singleton:
_instance = None
_lock = threading.Lock() # 类级别锁，保证线程安全
def __new__(cls, *args, **kwargs):
if not cls._instance: # 第一层检查，提高效率
with cls._lock: # 加锁，防止多线程同时创建
if not cls._instance: # 第二层检查，防止重复创建
cls._instance = super().__new__(cls)
return cls._instance
def __init__(self, value=None):
self.value = value
# 测试多线程
def task(name):
obj = Singleton(name)
print(f"{name} -> {obj.value}, id: {id(obj)}")
threads = []
for i in range(5):
t = threading.Thread(target=task, args=(f"Thread-{i}",))
threads.append(t)
t.start()
for t in threads:
t.join()
多核CPU
在多核 CPU 上，很多人会认为多个线程能够真正并行执行。那如果我们编写一个死循环线程，会发生什么呢？
通过 Mac OS X 的活动监视器或 Windows 的任务管理器观察，可以发现：一个死循环线程会占满一个 CPU 核心（CPU 使用率 100%）。如果同时运行两个死循环线程，在多核环境下就会占用两个核心（CPU 使用率 200%）。因此，要想让 N 核 CPU 的所有核心都跑满，就需要同时启动 N 个死循环线程
import threading, multiprocessing
def loop():
x = 0
while True:
x = x ^ 1
for i in range(multiprocessing.cpu_count()):
t = threading.Thread(target=loop)
t.start()
如果启动与 CPU 核心数量相同的 N 个线程，在 4 核 CPU 上观察，CPU 使用率可能只有约 102%，也就是只真正使用了一核
而用 C、C++ 或 Java 实现相同的死循环，则可以轻松把所有核心跑满：4 核跑到 400%，8 核跑到 800%。为什么 python 做不到呢？
原因在于 python 虽然有真正的线程，但解释器存在一个 GIL（Global Interpreter Lock，全局解释器锁）。任何 Python 线程在执行前都必须先获取 GIL，每执行大约 100 条字节码后，解释器才会释放 GIL，让其他线程有机会执行。
由于这个全局锁，python 的多线程只能交替运行，即使 100 个线程在 100 核 CPU 上，也只能用到一个核心 GIL 是 python 解释器设计的历史遗留问题，尤其是官方的 CPython。要真正利用多核，除非使用不带 GIL 的解释器或进行 C 扩展，否则 python 多线程无法有效利用多核
不过不用太担心：python 虽然不能用多线程充分利用多核，但可以通过多进程实现多核并行。每个 python 进程都有独立的 GIL，互不干扰，因此可以充分利用多核 CPU
ThreadLocal
在多线程环境中，每个线程都有自己的数据。使用线程的局部变量比使用全局变量更安全，因为局部变量只对该线程可见，不会影响其他线程；而全局变量的修改则必须加锁才能避免冲突
不过，局部变量也有缺点：在函数调用时，需要手动传递，使用起来比较麻烦
def process_student(name):
std = Student(name)
# std是局部变量，但是每个函数都要用它，因此必须传进去：
do_task_1(std)
do_task_2(std)
def do_task_1(std):
do_subtask_1(std)
do_subtask_2(std)
def do_task_2(std):
do_subtask_2(std)
do_subtask_2(std)
如果每个函数都一层层传递参数，那就太麻烦了。直接用全局变量也不行，因为每个线程处理的都是不同的 Student 对象，不能共享
有没有办法把所有 Student 对象放在一个全局字典里，然后用线程自身作为 key 来获取对应线程的 Student 对象呢？
global_dict = {}
def std_thread(name):
std = Student(name)
# 把std放到全局变量global_dict中：
global_dict[threading.current_thread()] = std
do_task_1()
do_task_2()
def do_task_1():
# 不传入std，而是根据当前线程查找：
std = global_dict[threading.current_thread()]
...
def do_task_2():
# 任何函数都可以查找出当前线程的std变量：
std = global_dict[threading.current_thread()]
...
这种方法理论上是可行的，它最大的优点是解决了 Student 对象在每层函数中传递的问题。不过，每个函数都需要去字典里获取对象，代码看起来有些冗长。有没有更简单的办法呢？这时，ThreadLocal 应运而生：它可以自动为每个线程管理数据，不需要手动查找字典
import threading
# 创建全局ThreadLocal对象:
local_school = threading.local()
def process_student():
# 获取当前线程关联的student:
std = local_school.student
print('Hello, %s (in %s)' % (std, threading.current_thread().name))
def process_thread(name):
# 绑定ThreadLocal的student:
local_school.student = name
process_student()
t1 = threading.Thread(target= process_thread, args=('Alice',), name='Thread-A')
t2 = threading.Thread(target= process_thread, args=('Bob',), name='Thread-B')
t1.start()
t2.start()
t1.join()
t2.join()
执行结果
import threading
# 创建全局ThreadLocal对象:
local_school = threading.local()
def process_student():
# 获取当前线程关联的student:
std = local_school.student
print('Hello, %s (in %s)' % (std, threading.current_thread().name))
def process_thread(name):
# 绑定ThreadLocal的student:
local_school.student = name
process_student()
t1 = threading.Thread(target= process_thread, args=('Alice',), name='Thread-A')
t2 = threading.Thread(target= process_thread, args=('Bob',), name='Thread-B')
t1.start()
t2.start()
t1.join()
t2.join()
全局变量 local_school 是一个 ThreadLocal 对象，每个线程都可以读写它的 student 属性，但互不干扰。
你可以把 local_school 看作一个全局变量，但它的属性（如 local_school.student）实际上是线程的局部变量，可以自由读写，不需要手动管理锁，ThreadLocal 会在内部处理这些细节
可以把 ThreadLocal 想象成一个特殊的字典，不仅可以用 local_school.student，还可以绑定其他变量，比如 local_school.teacher 等等
进程 vs. 线程
要实现多任务，通常会采用 Master-Worker 模式：Master 负责分配任务，Worker 负责执行任务。因此，多任务环境下一般是一个 Master 配合多个 Worker
多进程实现：主进程作为 Master，其他子进程作为 Worker
- 优点：稳定性高，一个子进程崩溃不会影响主进程或其他子进程（当然如果 Master 挂掉，所有进程都会受影响，但 Master 只负责分配任务，挂掉的概率低）
- 缺点：创建进程开销大。在 Unix/Linux 下用
fork开销还可以接受，但在 Windows 上创建进程代价很高。此外，操作系统能同时调度的进程数有限，如果同时运行几千个进程，操作系统甚至可能连调度都成问题
多线程实现：主线程作为 Master，其他线程作为 Worker
- 优点：通常比多进程稍快，但提升有限
- 缺点：危险性大，任何一个线程出错都可能导致整个进程崩溃，因为所有线程共享进程内存。在 Windows 上，如果某个线程执行异常，你经常会看到“该程序执行了非法操作，即将关闭”的提示，实际上只是某个线程出错，但操作系统会强制结束整个进程
线程切换
无论是多进程还是多线程，当数量一多，效率往往不会提升，为什么呢？
我们打个比方：假设你正在准备中考，每天晚上需要完成语文、数学、英语、物理、化学五科作业，每科作业耗时 1 小时
- 单任务模型：你先花 1 小时做语文作业，再花 1 小时做数学作业，依次完成所有作业，一共需要 5 小时
- 多任务模型：你尝试轮流做每科作业，每科做 1 分钟再切换到下一科，如此循环。只要切换够快，从表面上看，你似乎同时在做 5 科作业，就像单核 CPU 在执行多任务一样
问题是，切换作业是有代价的。比如从语文切换到数学，你要先收拾语文书、钢笔（保存现场），再打开数学课本、找出直尺圆规（准备新环境）。操作系统在切换进程或线程时也类似：需要先保存当前执行环境（CPU 寄存器、内存页等），再恢复新任务的环境，才能开始执行。
虽然切换很快，但仍然消耗时间
如果同时有几千个任务，操作系统可能主要忙于切换任务，而没有足够时间真正执行任务。结果就是，多任务一旦过多，就会耗尽系统资源，效率急剧下降，所有任务都完成得很慢
计算密集型 vs. IO密集型
在决定是否采用多任务时，另一个重要因素是任务类型。任务大体可分为计算密集型和 IO 密集型两类
- 计算密集型任务：主要消耗 CPU 资源，需要大量计算，例如计算圆周率、高清视频解码等。这类任务虽然可以使用多任务，但任务越多，花在切换任务上的时间也越多，CPU 执行效率反而下降。因此，为了最有效地利用 CPU，计算密集型任务同时运行的数量最好与 CPU 核心数相等。由于计算密集型任务依赖 CPU 性能，代码运行效率非常重要。像 python 这样的脚本语言运行速度较慢，不适合处理计算密集型任务；这类任务最好使用 C 语言编写
- IO 密集型任务：涉及网络或磁盘操作，CPU 消耗很少，大部分时间都在等待 IO 完成（因为 IO 速度远低于 CPU 和内存）。对于 IO 密集型任务，增加任务数量可以提升 CPU 利用率，但也有上限。常见的 Web 应用大多属于 IO 密集型任务。由于执行期间绝大部分时间花在等待 IO，使用运行速度更快的 C 语言替换 python 并不会显著提高效率。因此，对于 IO 密集型任务，选择开发效率高、代码量少的语言最合适，脚本语言是首选，而 C 语言开发效率相对较低
异步IO
由于 CPU 和 IO 速度差异巨大，任务在执行过程中大部分时间都在等待 IO，如果仍然使用单进程单线程模型，其他任务就无法并行执行。因此，我们才需要多进程或多线程来实现多任务并发
现代操作系统在 IO 方面做了大量优化，其中最重要的就是对异步 IO 的支持。如果充分利用异步 IO，即便是单进程单线程，也能高效执行多任务，这就是所谓的事件驱动模型。例如 Nginx 就是一款支持异步 IO 的 Web 服务器：在单核 CPU 上，它
用单进程就能高效处理大量请求；在多核 CPU 上，可以启动与核心数相等的进程，充分利用多核资源。由于系统总进程数有限，操作系统的调度也非常高效
在 python 中，单线程的异步编程模型称为协程
协程
可将协程想象成一个普通的 python 函数，但它具有一个超能力：在遇到可能需要一段时间才能完成的操作时，能够暂停执行。当长时间运行的操作完成时，可唤醒暂停的协程，并执行该协程中的其他代码。
当一个暂停的协程正在等待操作完成时，可运行其他代码，等待时其他代码的运行是应用程序并发的原因。还可同时运行多个耗时的操作，这能大大提高应用程序的性能。
当有一个长时间运行的操作时，await 关键字可以让我们暂停协程
关键字
async
async 关键字将函数标记为协程函数，而不是普通的 python 函数
async def coroutine():
print("hello world")
这是一个简单的协程函数，不执行任何长时间的操作，它只是输出信息并返回。将上述协程放在事件循环中时，它将立即执行，因为没有任何阻塞I/O，没有任何操作暂停执行
async def coroutine_add_one(number):
return number + 1
def add_one(number):
return number + 1
function_result = add_one(1)
coroutine_result = coroutine_add_one(1)
print(function_result)
print(type(function_result))
"""
2
<class 'int'>
"""
print(coroutine_result)
print(type(coroutine_result))
"""
<coroutine object coroutine_add_one at 0x000002977045BAC0>
<class 'coroutine'>
"""
调用普通的 add_one 函数时，它会立即执行并返回我们期望的一个整数。但当调用 coroutine_add_one 时，并不会执行协程中的代码，而是得到一个协程对象。当直接调用协程函数时，协程不会被执行。相反，它创建了一个可以稍后执行的协程对象，要执行协程，需要在事件循环中显式执行它。那么如何创建个事件循环并执行协程呢?
使用asyncio.run()函数来运行协程
import asyncio
async def coroutine_add_one(number):
return number + 1
coroutine_result = asyncio.run(coroutine_add_one(1))
print(coroutine_result) # 2
正如我们期望的一样，我们已经正确地将协程放在事件循环中，并且已经执行了它
asyncio.run 在这种情况下完成了一些重要的事情，首先创建了一个全新的事件循环。一旦成功创建，就会接受我们传递给它的任何协程，并运行它直到完成，然后返回结果，一切完成后，它会关闭并结束事件循环
await
asyncio的真正优势是能暂停执行，让事件循环在长时间运行的操作期间，运行其他任务
import asyncio
async def add_one(number):
return number + 1
async def main():
# main() 协程将暂停执行，直到 add_one(1) 运行完毕
one_plus_one = await add_one(1)
# main() 协程将暂停执行，直到 add_one(2) 运行完毕
two_plus_one = await add_one(2)
print(one_plus_one)
print(two_plus_one)
asyncio.run(main())
"""
2
3
"""
在上面的代码中，我们两次暂停执行。
首先等待对 add_one(1) 的调用，一旦得到结果，主函数将取消暂停并将 add_one(1) 的返回值分配给变量 one_plus_one。然后对 add_one(2) 执行相同的操作，并输出结果。我们来将应用程序的执行流程可视化一下，如下图所示，图中的每个块代表一行或多行代码在任何给定时刻发生的事情
asyncio.sleep
使用 asyncio.sleep 让协程休眠给定的秒数，这将在预定的时间内暂停协程。由于 asyncio.sleep 本身是一个协程，所以必须将它与 await 关键字一起使用，如果单独调用它，会得到一个协程对象。既然 asyncio.sleep 是一个协程，这意味着当协程等待它时，其他代码也能够运行
import asyncio
async def hello_world():
# 暂停 hello_world 协程一秒钟
await asyncio.sleep(1)
return "hello world"
async def main():
# 暂停 main 协程，直到 hello_world 协程运行完毕
message = await hello_world()
print(message)
asyncio.run(main())
"""
hello world
"""
运行这个应用程序时，程序将等待 1 秒钟，然后输出打印信息。由于 hello_world 是一个协程，使用 asyncio.sleep 将其暂停 1 秒，因此现在有 1 秒的时间可以同时运行其他代码
import asyncio
async def delay(seconds):
print(f"开始休眠 {seconds} 秒")
await asyncio.sleep(seconds)
print(f"休眠完成")
return seconds
async def add_one(number):
return number + 1
async def hello_world():
await delay(1)
return "hello world"
async def main():
# 暂停 main()，直到 add_one(1) 返回
one_plus_one = await add_one(1)
# 暂停 main()，直到 hello_world() 返回
message = await hello_world()
print(one_plus_one)
print(message)
asyncio.run(main())
"""
开始休眠 1 秒
休眠完成
2
hello world
"""
在 main 协程里面分别通过 await 驱动 add_one(1) 和 hello_world() 两个协程执行，然后打印它们的返回值，但是在打印 one_plus_one 之前需要等待一秒，因为在 hello_world() 协程里面 sleep 了一秒。但我们真正想要的结果是，在 await sleep 的时候，立刻执行其它的代码，比如立刻打印 one_plus_one，但实际情况却没有。
这是为什么呢？答案是在 await 暂停当前的协程之后、以及 await 表达式给我们一个值之前不会执行该协程中的其他任何代码。因为 hello_world_message 函数需要 1 秒后才能给出一个值，所以主协程将暂停 1 秒。这种情况下，代码表现得好像它是串行的
事实上从源代码本身也能够理解，因为代码是一行一行写的，所以自然也要一行一行执行。而 await 后面跟一个协程之后，会驱动协程执行，并等到驱动的协程运行完毕之后才往下执行。因此这个逻辑就决定了，await 是串行的，一个 await 执行完毕之后才能执行下一个 await。如果我们想摆脱这种顺序模型，同时运行 add_one 和 hello_world，那么需要引入一个被称为 “任务” 的概念
任务
直接调用协程时，并没有把它放在事件循环中运行，相反会得到一个协程对象。如果想运行，要么通过 asyncio.run，要么在一个协程里面通过 await 关键字进行驱动（在 A 协程里面 await B 协程，如果 A 协程运行了，那么 B 协程也会被驱动）
创建任务
创建任务是通过 asyncio.create_task 函数来实现的，当调用这个函数时，需要给它传递一个协程，然后返回一个任务对象。一旦有了一个任务对象，就可以把它放在一个 await 表达式中，它完成后就会提取返回值
import asyncio
async def delay(seconds):
print(f"开始休眠 {seconds} 秒")
await asyncio.sleep(seconds)
print(f"休眠完成")
return seconds
async def main():
# 将 delay(3) 包装成任务，注：包装完之后直接就丢到事件循环里面运行了
# 因此这里会立即返回，而返回值是一个 asyncio.Task 对象
sleep_for_three = asyncio.create_task(delay(3))
print("sleep_for_three:", sleep_for_three.__class__)
# 至于协程究竟有没有运行完毕，我们可以通过 Task 对象来查看
# 当协程运行完毕或者报错，都看做是运行完毕了，那么调用 Task 对象的 done 方法会返回 True
# 否则返回 False，由于代码是立即执行，还没有到 3 秒钟，因此打印结果为 False
print("协程(任务)是否执行完毕:", sleep_for_three.done())
# 这里则保证必须等到 Task 对象里面的协程运行完毕后，才能往下执行
result = await sleep_for_three
print("协程(任务)是否执行完毕:", sleep_for_three.done())
print("返回值:", result)
asyncio.run(main())
"""
开始休眠 3 秒
sleep_for_three: <class '_asyncio.Task'>
协程(任务)是否执行完毕: False
休眠完成
协程(任务)是否执行完毕: True
返回值: 3
"""
如果我们直接 await delay(3)，那么在打印之前需要至少等待 3 秒，但通过将它包装成任务，会立即扔到事件循环里面运行。此时主程序可以直接往下执行，至于协程到底什么时候执行完毕、有没有执行完毕，则通过 Task 对象（任务）来查看。当然你也可以 await 一个 Task 对象，保证里面的协程运行完毕后才能往下执行
同时运行多个任务
任务是立即创建并计划尽快运行，这允许同时运行许多长时间的任务
import asyncio
async def delay(seconds):
print(f"开始休眠 {seconds} 秒")
await asyncio.sleep(seconds)
print(f"休眠完成")
return seconds
async def main():
sleep_for_three = asyncio.create_task(delay(3))
sleep_again = asyncio.create_task(delay(3))
sleep_once_more = asyncio.create_task(delay(3))
await sleep_for_three
await sleep_again
await sleep_once_more
asyncio.run(main())
"""
开始休眠 3 秒
开始休眠 3 秒
开始休眠 3 秒
休眠完成
休眠完成
休眠完成
"""
在上面的代码中启动了三个任务，每个任务需要 3 秒才能完成。但由于对 create_task 的每次调用都会立即返回，因此会立即到达 await sleep_for_three语句，并且三个任务都丢到了事件循环，开启执行。由于 asyncio.sleep 属于 IO，因此会进行切换，所以三个任务是并发执行的，这也意味着整个程序会在 3 秒钟左右完成，而不是 9 秒钟
随着我们添加更多任务，性能提升效果会更明显，比如启动了 10 个这样的任务，仍然只需要大约 3 秒，从而使速度提高 10 倍
import asyncio
async def delay(seconds):
print(f"开始休眠 {seconds} 秒")
await asyncio.sleep(seconds)
print(f"休眠完成")
return seconds
async def hello_from_second():
for i in range(10):
await asyncio.sleep(1)
print("你好，我每秒钟负责打印一次")
async def main():
sleep_for_three = asyncio.create_task(delay(3))
sleep_again = asyncio.create_task(delay(3))
await hello_from_second()
asyncio.run(main())
"""
开始休眠 3 秒
开始休眠 3 秒
你好，我每秒钟负责打印一次
你好，我每秒钟负责打印一次
休眠完成
休眠完成
你好，我每秒钟负责打印一次
你好，我每秒钟负责打印一次
你好，我每秒钟负责打印一次
你好，我每秒钟负责打印一次
你好，我每秒钟负责打印一次
你好，我每秒钟负责打印一次
你好，我每秒钟负责打印一次
你好，我每秒钟负责打印一次
"""
在实际工作中，不要直接 await 一个协程，而是将协程包装成任务来让它运行。当你的代码逻辑依赖某个任务的执行结果时，再对该任务执行 await，拿到它的返回值
取消任务和设置超时
网络连接可能不可靠，用户的连接可能因为网速变慢而中断，或者网络服务器崩溃导致现有的请求无法处理。因此对于发出的请求，需要特别小心，不要无限期地等待。如果无限期等待一个不会出现的结果，可能导致应用程序挂起，从而导致精糕的用户体验在之前的示例中，如果任务一直持续下去，我们将被困在等待 await 语句完成而没有反馈的情况，也没有办法阻止这样的事情发生。因此 asyncio 提供了一个机制，允许我们手动取消任务，或者超时之后自动取消
取消任务
取消任务很简单，每个任务对象都有一个名为 cancel 的方法，可以在想要停止任务时调用它。取消一个任务将导致该任务在执行 await 时引发 CancelledError，然后再根据需要处理它
为说明这一点，假设启动了一个长时间运行的任务，但我们不希望它运行的时间超过 5 秒。如果任务没有在 5 秒内完成，就可以停止该任务，并向用户报告：该任务花费了太长时间，我们正在停止它。我们还希望每秒钟都输出一个状态更新，为用户提供最新信息，这样就可以让用户了解任务的运行状态
import asyncio
async def delay(seconds):
print(f"开始休眠 {seconds} 秒")
await asyncio.sleep(seconds)
print(f"休眠完成")
return seconds
async def main():
long_task = asyncio.create_task(delay(10))
seconds_elapsed = 0
while not long_task.done():
print("检测到任务尚未完成，一秒钟之后继续检测")
await asyncio.sleep(1)
seconds_elapsed += 1
# 时间超过 5 秒，取消任务
if seconds_elapsed == 5:
long_task.cancel()
try:
# 等待 long_task 完成，显然执行到这里的时候，任务已经被取消
# 不管是 await 一个已经取消的任务，还是 await 的时候任务被取消
# 都会引发 asyncio.CancelledError
await long_task
except asyncio.CancelledError:
print("任务被取消")
asyncio.run(main())
"""
检测到任务尚未完成，一秒钟之后继续检测
开始休眠 10 秒
检测到任务尚未完成，一秒钟之后继续检测
检测到任务尚未完成，一秒钟之后继续检测
检测到任务尚未完成，一秒钟之后继续检测
检测到任务尚未完成，一秒钟之后继续检测
检测到任务尚未完成，一秒钟之后继续检测
任务被取消
"""
在代码中我们创建了一个任务，它需要花费 10 秒的时间才能运行完成。然后创建一个 while 循环来检查该任务是否已完成，任务的 done 方法在任务完成时返回 True，否则返回 False。每一秒，我们检查任务是否已经完成，并记录到目前为止经历了多少秒。如果任务已经花费了 5 秒，就取消这个任务。然后来到 await long_task，将输出 “任务被取消”，这表明捕获了一个 CancelledError。
关于取消任务需要注意的是，CancelledError 只能从 await 语句抛出
import asyncio
async def delay(seconds):
print(f"开始休眠 {seconds} 秒")
await asyncio.sleep(seconds)
print(f"休眠完成")
return seconds
async def main():
long_task = asyncio.create_task(delay(3))
# 立刻取消
long_task.cancel()
# 但 CancelledError 只有在 await 取消的协程时才会触发
# 所以下面的语句会正常执行
print("我会正常执行")
print("Hello World")
print(list(range(10)))
await asyncio.sleep(5)
try:
# 引发 CancelledError
await long_task
except asyncio.CancelledError:
print("任务被取消")
asyncio.run(main())
"""
我会正常执行
Hello World
[0, 1, 2, 3, 4, 5, 6, 7, 8, 9]
任务被取消
"""
注意：如果任务在取消的时候已经运行完毕了，那么 await 的时候就不会抛 CancelledError 了
import asyncio
async def delay(seconds):
print(f"开始休眠 {seconds} 秒")
await asyncio.sleep(seconds)
print(f"休眠完成")
return seconds
async def main():
long_task = asyncio.create_task(delay(3))
await asyncio.sleep(5)
# 显然执行到这里，任务已经结束了
long_task.cancel()
try:
await long_task
print("任务执行完毕")
except asyncio.CancelledError:
print("任务被取消")
asyncio.run(main())
"""
开始休眠 3 秒
休眠完成
任务执行完毕
"""
所以对一个已完成的任务调用 cancel 方法，没有任何影响
设置超时并使用wait_for执行取消
每秒（或其他时间间隔）执行检查然后取消任务，并不是处理超时的最简单方法。理想情况下，我们应该有一个辅助函数，它允许指定超时并自动取消任务。
asyncio 通过名为 asyncio.wait_for 的函数提供此功能，该函数接收协程或任务对象，以及以秒为单位的超时时间。如果任务完成所需的时间超过了设定的超时时间，则会引发 TimeoutException，任务将自动取消为说明 wait_for 的工作原理，我们使用一个案例来说明：有一个任务需要 2 秒才能完成，但我们将它的超时时间设定为 1 秒。当得到一个 TimeoutError 异常时，我们将捕获异常，并检查任务是否被取消
import asyncio
async def delay(seconds):
print(f"开始休眠 {seconds} 秒")
await asyncio.sleep(seconds)
print(f"休眠完成")
return seconds
async def main():
delay_task = asyncio.create_task(delay(2))
try:
result = await asyncio.wait_for(delay_task, 1)
print("返回值:", result)
except asyncio.TimeoutError:
print("超时啦")
# delay_task.cancelled() 用于判断任务是否被取消
# 任务被取消：返回 True，没有被取消：返回 False
print("任务是否被取消:", delay_task.cancelled())
asyncio.run(main())
"""
开始休眠 2 秒
超时啦
任务是否被取消: True
"""
应用程序运行 1 秒后，wait_for 语句将引发 TimeoutError，然后我们对其进行处理，并且 delay_task 被取消了。所以当一个任务超时的时候，会被自动取消
所以通过 wait_for 语句就很方便，如果直接 await 一个任务，那么必须等到任务完成之后才能继续往下执行。
如果任务一直完成不了，那么就会一直陷入阻塞。我们的目的是希望这个任务的执行时间是可控的，那么便可以使用 wait_for 并指定超时时间。
注：使用 wait_for 必须要搭配 await，阻塞等待任务完成并拿到返回值、或者达到超时时间引发 TimeoutError 之后，程序才能往下执行因此 “await 任务” 和 “await asyncio.wait_for(任务, timeout)” 的效果是类似的，都是等待后面的任务完成并拿到它的返回值。但使用 wait_for 可以指定超时时间，在规定时间内如果没有完成，则抛出 TimeoutError，而不会一直陷入阻塞
如果任务花费的时间比预期的长，在引发 TimeoutError 之后自动取消任务通常是个好主意。否则，可能有一个协程无限期地等待，占用永远不会释放的资源。但在某些情况下，我们可能希望保持协程运行。
例如，我们可能想通知用户：某任务花费的时间比预期的要长，但即便超过了规定的超时时间，也不取消该任务。为此，可使用 asyncio.shield 函数包装任务，这个函数将防止传入的协程被取消，会给它一个屏蔽，将取消请求将忽略掉
import asyncio
async def delay(seconds):
print(f"开始休眠 {seconds} 秒")
await asyncio.sleep(seconds)
print(f"休眠完成")
return seconds
async def main():
delay_task = asyncio.create_task(delay(2))
try:
# 通过 asyncio.shield 将 delay_task 保护起来
result = await asyncio.wait_for(asyncio.shield(delay_task), 1)
print("返回值:", result)
except asyncio.TimeoutError:
print("超时啦")
# 如果超时依旧会引发 TimeoutError，但和之前不同的是
# 此时任务不会被取消了，因为 asyncio.shield 会将取消请求忽略掉
print("任务是否被取消:", delay_task.cancelled())
# 从出现超时的地方，继续执行，并等待它完成
result = await delay_task
print("返回值:", result)
asyncio.run(main())
"""
开始休眠 2 秒
超时啦
任务是否被取消: False
休眠完成
返回值: 2
"""
总结
以上为个人经验，希望能给大家一个参考，也希望大家多多支持脚本之家。