物无定味适口者珍,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密集型任务行不行?也行,但依然没有任何额外的收益, 无他,唯物无定味,适口者珍矣。

相关文章
|
2月前
|
消息中间件 调度
如何区分进程、线程和协程?看这篇就够了!
本课程主要探讨操作系统中的进程、线程和协程的区别。进程是资源分配的基本单位,具有独立性和隔离性;线程是CPU调度的基本单位,轻量且共享资源,适合并发执行;协程更轻量,由程序自身调度,适合I/O密集型任务。通过学习这些概念,可以更好地理解和应用它们,以实现最优的性能和资源利用。
81 11
|
4月前
|
API 开发者 Python
探索Python中的异步编程:Asyncio与Tornado的对决
在这个快节奏的世界里,Python开发者面临着一个挑战:如何让代码跑得更快?本文将带你走进Python异步编程的两大阵营——Asyncio和Tornado,探讨它们如何帮助我们提升性能,以及在实际应用中如何选择。我们将通过一场虚拟的“对决”,比较这两个框架的性能和易用性,让你在异步编程的战场上做出明智的选择。
|
4月前
|
测试技术 Python
Python中的异步编程与`asyncio`库
Python中的异步编程与`asyncio`库
|
4月前
|
API 调度 开发者
探索Python中的异步编程:从asyncio到Trio
本文将带你深入Python异步编程的心脏地带,从asyncio的基本概念到Trio的高级特性,我们将一起揭开Python异步编程的神秘面纱,并探讨它们如何改变我们的编程方式。
|
4月前
|
Python
Python中的异步编程:使用asyncio和aiohttp实现高效网络请求
【10月更文挑战第34天】在Python的世界里,异步编程是提高效率的利器。本文将带你了解如何使用asyncio和aiohttp库来编写高效的网络请求代码。我们将通过一个简单的示例来展示如何利用这些工具来并发地处理多个网络请求,从而提高程序的整体性能。准备好让你的Python代码飞起来吧!
176 2
|
4月前
|
数据库 Python
异步编程不再难!Python asyncio库实战,让你的代码流畅如丝!
在编程中,随着应用复杂度的提升,对并发和异步处理的需求日益增长。Python的asyncio库通过async和await关键字,简化了异步编程,使其变得流畅高效。本文将通过实战示例,介绍异步编程的基本概念、如何使用asyncio编写异步代码以及处理多个异步任务的方法,帮助你掌握异步编程技巧,提高代码性能。
133 4
|
4月前
|
API 数据处理 Python
探秘Python并发新世界:asyncio库,让你的代码并发更优雅!
在Python编程中,随着网络应用和数据处理需求的增长,并发编程变得愈发重要。asyncio库作为Python 3.4及以上版本的标准库,以其简洁的API和强大的异步编程能力,成为提升性能和优化资源利用的关键工具。本文介绍了asyncio的基本概念、异步函数的定义与使用、并发控制和资源管理等核心功能,通过具体示例展示了如何高效地编写并发代码。
71 2
|
4月前
|
调度 开发者 Python
Python中的异步编程:理解asyncio库
在Python的世界里,异步编程是一种高效处理I/O密集型任务的方法。本文将深入探讨Python的asyncio库,它是实现异步编程的核心。我们将从asyncio的基本概念出发,逐步解析事件循环、协程、任务和期货的概念,并通过实例展示如何使用asyncio来编写异步代码。不同于传统的同步编程,异步编程能够让程序在等待I/O操作完成时释放资源去处理其他任务,从而提高程序的整体效率和响应速度。
|
4月前
|
NoSQL 关系型数据库 MySQL
python协程+异步总结!
本文介绍了Python中的协程、asyncio模块以及异步编程的相关知识。首先解释了协程的概念和实现方法,包括greenlet、yield关键字、asyncio装饰器和async/await关键字。接着详细讲解了协程的意义和应用场景,如提高IO密集型任务的性能。文章还介绍了事件循环、Task对象、Future对象等核心概念,并提供了多个实战案例,包括异步Redis、MySQL操作、FastAPI框架和异步爬虫。最后提到了uvloop作为asyncio的高性能替代方案。通过这些内容,读者可以全面了解和掌握Python中的异步编程技术。
71 0
|
4月前
|
数据采集 缓存 程序员
python协程使用教程
1. **协程**:介绍了协程的概念、与子程序的区别、优缺点,以及如何在 Python 中使用协程。 2. **同步与异步**:解释了同步与异步的概念,通过示例代码展示了同步和异步处理的区别和应用场景。 3. **asyncio 模块**:详细介绍了 asyncio 模块的概述、基本使用、多任务处理、Task 概念及用法、协程嵌套与返回值等。 4. **aiohttp 与 aiofiles**:讲解了 aiohttp 模块的安装与使用,包括客户端和服务器端的简单实例、URL 参数传递、响应内容读取、自定义请求等。同时介绍了 aiofiles 模块的安装与使用,包括文件读写和异步迭代
80 0

推荐镜像

更多