物无定味适口者珍,Python3并发场景(CPU密集/IO密集)任务的并发方式的场景抉择(多线程threading/多进程multiprocessing/协程asyncio)

简介: 一般情况下,大家对Python原生的并发/并行工作方式:进程、线程和协程的关系与区别都能讲清楚。甚至具体的对象名称、内置方法都可以如数家珍,这显然是极好的,但我们其实都忽略了一个问题,就是具体应用场景,三者的使用目的是一样的,换句话说,使用结果是一样的,都可以提高程序运行的效率,但到底那种场景用那种方式更好一点?

一般情况下,大家对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:

pip3 install 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次,需要花费多少时间?

4.466595888137817秒

再来试试并行方式:

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}秒")

八核处理器,开八个进程开始跑:

1.1755797863006592秒

不言而喻,并行方式有效提高了计算效率。

最后,既然之前用并行方式运行了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个线程,走起:

4.452666759490967秒

如何?和并行方式运行IO密集型任务一样,可以运行,但是没有任何意义。为什么?因为没有任何IO操作了,CPU不需要等待了,CPU只要全力运算即可,所以你上多线程或者协程,无非就是画蛇添足、多此一举。

结语

有经验的汽修师傅会告诉你,想省油就选CVT和双离合,想质量稳定就选AT,经常高速上激烈驾驶就选双离合,经常市区内堵车就选CVT;同样地,作为经验丰富的后台研发,你也可以告诉汽修师傅,任何不需要CPU等待的任务就选择并行(multiprocessing)的处理方式,而需要CPU等待时间过长的任务,选择并发(threading/asyncio)。反过来,我就想用CVT在高速上飙车,用双离合在市区堵车,行不行?行,但没有意义,或者说的更准确一些,没有任何额外的收益;而用并发方式执行CPU密集型任务,用并行方式执行IO密集型任务行不行?也行,但依然没有任何额外的收益, 无他,唯物无定味,适口者珍矣。

相关文章
|
6月前
|
数据采集 监控 调度
干货分享“用 多线程 爬取数据”:单线程 + 协程的效率反超 3 倍,这才是 Python 异步的正确打开方式
在 Python 爬虫中,多线程因 GIL 和切换开销效率低下,而协程通过用户态调度实现高并发,大幅提升爬取效率。本文详解协程原理、实战对比多线程性能,并提供最佳实践,助你掌握异步爬虫核心技术。
|
消息中间件 并行计算 安全
进程、线程、协程
【10月更文挑战第16天】进程、线程和协程是计算机程序执行的三种基本形式。进程是操作系统资源分配和调度的基本单位,具有独立的内存空间,稳定性高但资源消耗大。线程是进程内的执行单元,共享内存,轻量级且并发性好,但同步复杂。协程是用户态的轻量级调度单位,适用于高并发和IO密集型任务,资源消耗最小,但不支持多核并行。
304 1
|
12月前
|
消息中间件 调度
如何区分进程、线程和协程?看这篇就够了!
本课程主要探讨操作系统中的进程、线程和协程的区别。进程是资源分配的基本单位,具有独立性和隔离性;线程是CPU调度的基本单位,轻量且共享资源,适合并发执行;协程更轻量,由程序自身调度,适合I/O密集型任务。通过学习这些概念,可以更好地理解和应用它们,以实现最优的性能和资源利用。
390 11
|
存储 缓存 算法
面试官:单核 CPU 支持 Java 多线程吗?为什么?被问懵了!
本文介绍了多线程环境下的几个关键概念,包括时间片、超线程、上下文切换及其影响因素,以及线程调度的两种方式——抢占式调度和协同式调度。文章还讨论了减少上下文切换次数以提高多线程程序效率的方法,如无锁并发编程、使用CAS算法等,并提出了合理的线程数量配置策略,以平衡CPU利用率和线程切换开销。
面试官:单核 CPU 支持 Java 多线程吗?为什么?被问懵了!
|
存储 消息中间件 人工智能
进程,线程,协程 - 你了解多少?
本故事采用简洁明了的对话方式,尽洪荒之力让你在轻松无负担的氛围中,稍微深入地理解进程、线程和协程的相关原理知识
138 2
进程,线程,协程 - 你了解多少?
|
消息中间件 并行计算 安全
进程、线程、协程
【10月更文挑战第15天】进程、线程和协程是操作系统中三种不同的执行单元。进程是资源分配和调度的基本单位,每个进程有独立的内存空间;线程是进程内的执行路径,共享进程资源,切换成本较低;协程则更轻量,由用户态调度,适合处理高并发和IO密集型任务。进程提供高隔离性和安全性,线程支持高并发,协程则在资源消耗和调度灵活性方面表现优异。
304 2
|
5月前
|
Go 调度 Python
Golang协程和Python协程用法上的那些“不一样”
本文对比了 Python 和 Go 语言中协程的区别,重点分析了调度机制和执行方式的不同。Go 的协程(goroutine)由运行时自动调度,启动后立即执行;而 Python 协程需通过 await 显式调度,依赖事件循环。文中通过代码示例展示了两种协程的实际运行效果。
238 7
|
4月前
|
数据采集 网络协议 API
协程+连接池:高并发Python爬虫的底层优化逻辑
协程+连接池:高并发Python爬虫的底层优化逻辑
|
Go Python
使用python实现一个用户态协程
【6月更文挑战第28天】本文探讨了如何在Python中实现类似Golang中协程(goroutines)和通道(channels)的概念。文章最后提到了`wait_for`函数在处理超时和取消操作中的作
252 1
使用python实现一个用户态协程
|
数据库 开发者 Python
实战指南:用Python协程与异步函数优化高性能Web应用
【7月更文挑战第15天】Python的协程与异步函数优化Web性能,通过非阻塞I/O提升并发处理能力。使用aiohttp库构建异步服务器,示例代码展示如何处理GET请求。异步处理减少资源消耗,提高响应速度和吞吐量,适用于高并发场景。掌握这项技术对提升Web应用性能至关重要。
296 10

推荐镜像

更多