原文转载自「刘悦的技术博客」https://v3u.cn/a_id_221
一般情况下，大家对 Python 原生的并发/并行工作方式：进程、线程和协程的关系与区别都能讲清楚。甚至具体的对象名称、内置方法都可以如数家珍，这显然是极好的，但我们其实都忽略了一个问题，就是具体应用场景，三者的使用目的是一样的，话句话说，使用结果是一样的，都可以提高程序运行的效率，但到底那种场景用那种方式更好一点？
这就好比，目前主流的汽车发动机变速箱无外乎三种：双离合、CVT 以及传统 AT。主机厂把它们搭载到不同的发动机和车型上，它们都是变速箱，都可以将发动机产生的动力作用到车轮上，但不同使用场景下到底该选择那种变速箱？这显然也是一个问题。
所谓“无场景，不功能”，本次我们来讨论一下，具体的并发编程场景有哪些，并且对应到具体场景，应该怎么选择并发手段和方式。
什么是并发和并行？
在讨论场景之前，我们需要将多任务执行的方式进行一下分类，那就是并发方式和并行方式。教科书上告诉我们：并行是指两个或者多个事件在同一时刻发生；而并发是指两个或多个事件在同一时间间隔内发生。 在多道程序环境下，并发性是指在一段时间内宏观上有多个程序在同时运行，但在单处理机系统中，每一时刻却仅能有一道程序执行，故微观上这些程序只能是分时地交替执行。
好像有那么一点抽象，好吧，让我们务实一点，由于 GIL 全局解释器锁的存在，在 Python 编程领域，我们可以简单粗暴地将并发和并行用程序通过能否使用多核 CPU 来区分，能使用多核 CPU 就是并行，不能使用多核 CPU，只能单核处理的，就是并发。就这么简单，是的，Python 的 GIL 全局解释器锁帮我们把问题简化了， 这是 Python 的大幸？还是不幸？
Python 中并发任务实现方式包含：多线程 threading 和协程 asyncio，它们的共同点都是交替执行，而区别是多线程 threading 是抢占式的，而协程 asyncio 是协作式的，原理也很简单，只有一颗 CPU 可以用，而一颗 CPU 一次只能做一件事，所以只能靠不停地切换才能完成并发任务。
Python 中并行任务的实现方式是多进程 multiprocessing，通过 multiprocessing 库，Python 可以在程序主进程中创建新的子进程。这里的一个进程可以被认为是一个几乎完全不同的程序，尽管从技术上讲，它们通常被定义为资源集合，其中资源包括内存、文件句柄等。换一种说法是，每个子进程都拥有自己的 Python 解释器，因此，Python 中的并行任务可以使用一颗以上的 CPU，每一颗 CPU 都可以跑一个进程，是真正的同时运行，而不需要切换，如此 Python 就可以完成并行任务。
什么时候使用并发？IO 密集型任务
现在我们搞清楚了，Python 里的并发运行方式就是多线程 threading 和协程 asyncio，那么什么场景下使用它们？
一般情况下，任务场景，或者说的更准确一些，任务类型，无非两种：CPU 密集型任务和 IO 密集型任务。
什么是 IO 密集型任务？IO 就是 Input-Output 的缩写，说白了就是程序的输入和输出，想一想确实就是这样，您的电脑，它不就是这两种功能吗？用键盘、麦克风、摄像头输入数据，然后再用屏幕和音箱进行输出操作。
但输入和输出操作要比电脑中的 CPU 运行速度慢，换句话说，CPU 得等着这些比它慢的输入和输出操作，说白了就是 CPU 运算一会，就得等这些 IO 操作，等 IO 操作完了，CPU 才能继续运算一会，然后再等着 IO 操作，如图所示：
由此可知，并发适合这种 IO 操作密集和频繁的工作，因为就算 CPU 是苹果最新 ARM 架构的 M2 芯片，也没有用武之地。
另外，如果把 IO 密集型任务具象化，那就是我们经常操作的：硬盘读写(数据库读写)、网络请求、文件的打印等等。
并发方式的选择：多线程 threading 还是协程 asyncio？
既然涉及硬盘读写(数据库读写)、网络请求、文件打印等任务都算并发任务，那我们就真正地实践一下，看看不同的并发方式到底能提升多少效率？
一个简单的小需求，对本站数据进行重复抓取操作，并计算首页数据文本的行数：
import requests import time def download_site(url, session): with session.get(url) as response: print(f"下载了{len(response.content)}行数据") def download_all_sites(sites): with requests.Session() as session: for url in sites: download_site(url, session) if __name__ == "__main__": sites = ["https://v3u.cn"] * 50 start_time = time.time() download_all_sites(sites) duration = time.time() - start_time print(f"下载了 {len(sites)}次，执行了{duration}秒")
复制代码
在不使用任何并发手段的前提下，程序返回：
下载了76347行数据 下载了76347行数据 下载了76347行数据 下载了76347行数据 下载了76347行数据 下载了76347行数据 下载了76347行数据 下载了76347行数据 下载了76347行数据 下载了76347行数据 下载了76347行数据 下载了76347行数据 下载了76347行数据 下载了76347行数据 下载了76347行数据 下载了76347行数据 下载了76347行数据 下载了76347行数据 下载了76347行数据 下载了76347行数据 下载了76347行数据 下载了76347行数据 下载了76347行数据 下载了76347行数据 下载了76347行数据 下载了76347行数据 下载了76347行数据 下载了76347行数据 下载了76347行数据 下载了76347行数据 下载了76347行数据 下载了76347行数据 下载了76347行数据 下载了76347行数据 下载了76347行数据 下载了76347行数据 下载了76347行数据 下载了76347行数据 下载了76347行数据 下载了76347行数据 下载了76347行数据 下载了76347行数据 下载了76347行数据 下载了76347行数据 下载了76347行数据 下载了76347行数据 下载了76347行数据 下载了76347行数据 下载了76347行数据 下载了76347行数据 下载了 50 次数据，执行了8.781155824661255秒 [Finished in 9.6s]
复制代码
这里程序的每一步都是同步操作，也就是说当第一次抓取网站首页时，剩下的 49 次都在等待。
接着使用多线程 threading 来改造程序：
import concurrent.futures import requests import threading import time thread_local = threading.local() def get_session(): if not hasattr(thread_local, "session"): thread_local.session = requests.Session() return thread_local.session def download_site(url): session = get_session() with session.get(url) as response: print(f"下载了{len(response.content)}行数据") def download_all_sites(sites): with concurrent.futures.ThreadPoolExecutor(max_workers=8) as executor: executor.map(download_site, sites) if __name__ == "__main__": sites = ["https://v3u.cn"] * 50 start_time = time.time() download_all_sites(sites) duration = time.time() - start_time print(f"下载了 {len(sites)}次，执行了{duration}秒")
复制代码
这里通过 with 关键词开启线程池上下文管理器，并发 8 个线程进行下载，程序返回：
下载了76424行数据 下载了76424行数据 下载了76424行数据 下载了76424行数据 下载了76424行数据 下载了76424行数据 下载了76424行数据 下载了76424行数据 下载了76424行数据 下载了76424行数据 下载了76424行数据 下载了76424行数据 下载了76424行数据 下载了76424行数据 下载了76424行数据 下载了76424行数据 下载了76424行数据 下载了76424行数据 下载了76424行数据 下载了76424行数据 下载了76424行数据 下载了76424行数据 下载了76424行数据 下载了76424行数据 下载了76424行数据 下载了76424行数据 下载了76424行数据 下载了76424行数据 下载了76424行数据 下载了76424行数据 下载了76424行数据 下载了76424行数据 下载了76424行数据 下载了76424行数据 下载了76424行数据 下载了76424行数据 下载了76424行数据 下载了76424行数据 下载了76161行数据 下载了76424行数据 下载了76161行数据 下载了76161行数据 下载了76161行数据 下载了76161行数据 下载了76161行数据 下载了76161行数据 下载了76161行数据 下载了76161行数据 下载了76161行数据 下载了76161行数据 下载了 50次，执行了7.680492877960205秒
复制代码
很明显，效率上有所提升，事实上，每个线程其实是在不停“切换”着运行，这就节省了单线程每次等待爬取结果的时间：
由此带来了另外一个问题：上下文切换的时间开销。
让我们继续改造，用协程来一试锋芒，首先安装异步 web 请求库 aiohttp:
改写逻辑：
import asyncio import time import aiohttp async def download_site(session, url): async with session.get(url) as response: print(f"下载了{response.content_length}行数据") async def download_all_sites(sites): async with aiohttp.ClientSession() as session: tasks = [] for url in sites: task = asyncio.ensure_future(download_site(session, url)) tasks.append(task) await asyncio.gather(*tasks, return_exceptions=True) if __name__ == "__main__": sites = ["https://v3u.cn"] * 50 start_time = time.time() asyncio.run(download_all_sites(sites)) duration = time.time() - start_time print(f"下载了 {len(sites)}次，执行了{duration}秒")
复制代码
程序返回：
下载了76424行数据 下载了76424行数据 下载了76424行数据 下载了76424行数据 下载了76424行数据 下载了76424行数据 下载了76424行数据 下载了76424行数据 下载了76424行数据 下载了76424行数据 下载了76424行数据 下载了76424行数据 下载了76424行数据 下载了76424行数据 下载了76424行数据 下载了76424行数据 下载了76424行数据 下载了76424行数据 下载了76424行数据 下载了76424行数据 下载了76424行数据 下载了76424行数据 下载了76424行数据 下载了76424行数据 下载了76424行数据 下载了76424行数据 下载了76424行数据 下载了76424行数据 下载了76424行数据 下载了76424行数据 下载了76424行数据 下载了76424行数据 下载了76424行数据 下载了76424行数据 下载了76424行数据 下载了76424行数据 下载了76424行数据 下载了76424行数据 下载了76161行数据 下载了76424行数据 下载了76161行数据 下载了76161行数据 下载了76161行数据 下载了76161行数据 下载了76161行数据 下载了76161行数据 下载了76161行数据 下载了76161行数据 下载了76161行数据 下载了76161行数据
下载了 50次，执行了6.893810987472534秒
复制代码
效率上百尺竿头更进一步，同样的使用 with 关键字操作上下文管理器，协程使用 asyncio.ensure_future()创建任务列表，该列表还负责启动它们。创建所有任务后，使用 asyncio.gather()来保持会话上下文的实例，直到所有爬取任务完成。和多线程 threading 的区别是，协程并不需要切换上下文，因此每个任务所需的资源和创建时间要少得多，因此创建和运行更多的任务效率更高：
综上，并发逻辑归根结底是减少 CPU 等待的时间，也就是让 CPU 少等一会儿，而协程的工作方式显然让 CPU 等待的时间最少。
并行方式：多进程 multiprocessing
再来试试多进程 multiprocessing，并行能不能干并发的事？
import requests import multiprocessing import time session = None def set_global_session(): global session if not session: session = requests.Session() def download_site(url): with session.get(url) as response: name = multiprocessing.current_process().name print(f"读了{len(response.content)}行") def download_all_sites(sites): with multiprocessing.Pool(initializer=set_global_session) as pool: pool.map(download_site, sites) if __name__ == "__main__": sites = ["https://v3u.cn"] * 50 start_time = time.time() download_all_sites(sites) duration = time.time() - start_time print(f"下载了 {len(sites)}次，执行了{duration}秒")
复制代码
这里我们依然使用上下文管理器开启进程池，默认进程数匹配当前计算机的 CPU 核心数，也就是有几核就开启几个进程，程序返回：
读了76000行 读了76241行 读了76044行 读了75894行 读了76290行 读了76312行 读了76419行 读了76753行 读了76290行 读了76290行 读了76290行 读了76290行 读了76290行 读了76290行 读了76290行 读了76290行 读了76290行 读了76290行 读了76290行 读了76290行 读了76290行 读了76290行 读了76290行 读了76290行 读了76290行 读了76290行 读了76290行 读了76290行 读了76290行 读了76290行 读了76290行 读了76290行 读了76290行 读了76290行 读了76290行 读了76290行 读了76290行 读了76290行 读了76290行 读了76290行 读了76290行 读了76290行 读了76290行 读了76290行 读了76290行 读了76290行 读了76290行 读了76290行 读了76290行 读了76290行 下载了 50次，执行了8.195281982421875秒
复制代码
虽然比同步程序要快，但无疑的，效率上要低于多线程和协程。为什么？因为多进程不适合 IO 密集型任务，虽然可以利用多核资源，但没有任何意义：
无论开多少进程，CPU 都没有用武之地，多数情况下 CPU 都在等待 IO 操作，也就是说，多核反而拖累了 IO 程序的执行。
并行方式的选择：CPU 密集型任务
什么是 CPU 密集型任务？这里我们可以使用逆定理：所有不涉及硬盘读写(数据库读写)、网络请求、文件打印等任务都算 CPU 密集型任务任务，说白了就是，计算型任务。
以求平方和为例子：
import time def cpu_bound(number): return sum(i * i for i in range(number)) def find_sums(numbers): for number in numbers: cpu_bound(number) if __name__ == "__main__": numbers = [5_000_000 + x for x in range(20)] start_time = time.time() find_sums(numbers) duration = time.time() - start_time print(f"{duration}秒")
复制代码
同步执行 20 次，需要花费多少时间？
再来试试并行方式：
import multiprocessing import time def cpu_bound(number): return sum(i * i for i in range(number)) def find_sums(numbers): with multiprocessing.Pool() as pool: pool.map(cpu_bound, numbers) if __name__ == "__main__": numbers = [5_000_000 + x for x in range(20)] start_time = time.time() find_sums(numbers) duration = time.time() - start_time print(f"{duration}秒")
复制代码
八核处理器，开八个进程开始跑：
不言而喻，并行方式有效提高了计算效率。
最后，既然之前用并行方式运行了 IO 密集型任务，我们就再来试试用并发的方式运行 CPU 密集型任务：
import concurrent.futures import time def cpu_bound(number): return sum(i * i for i in range(number)) def find_sums(numbers): with concurrent.futures.ThreadPoolExecutor(max_workers=8) as executor: executor.map(cpu_bound, numbers) if __name__ == "__main__": numbers = [5_000_000 + x for x in range(20)] start_time = time.time() find_sums(numbers) duration = time.time() - start_time print(f"{duration}秒")
复制代码
单进程开 8 个线程，走起：
如何？和并行方式运行 IO 密集型任务一样，可以运行，但是没有任何意义。为什么？因为没有任何 IO 操作了，CPU 不需要等待了，CPU 只要全力运算即可，所以你上多线程或者协程，无非就是画蛇添足、多此一举。
结语
有经验的汽修师傅会告诉你，想省油就选 CVT 和双离合，想质量稳定就选 AT，经常高速上激烈驾驶就选双离合，经常市区内堵车就选 CVT；同样地，作为经验丰富的后台研发，你也可以告诉汽修师傅，任何不需要 CPU 等待的任务就选择并行(multiprocessing)的处理方式，而需要 CPU 等待时间过长的任务，选择并发(threading/asyncio)。反过来，我就想用 CVT 在高速上飙车，用双离合在市区堵车，行不行？行，但没有意义，或者说的更准确一些，没有任何额外的收益；而用并发方式执行 CPU 密集型任务，用并行方式执行 IO 密集型任务行不行？也行，但依然没有任何额外的收益， 无他，唯物无定味，适口者珍矣。
原文转载自「刘悦的技术博客」 https://v3u.cn/a_id_221
评论