Python中asyncio的多种用法：异步同步

Python库中的asyncio

asyncio: 轻松实现异步并发任务

1 引言 📖

Python 的 asyncio 模块为异步编程提供了强大的支持，但在某些场景下，我们可能需要处理异步任务与非异步（同步）任务的顺序执行或并行执行。本篇文章将逐步带你了解如何在 Python 中处理这些不同类型的任务。

2 顺序执行非异步任务 🕒

在日常编程中，最常见的情况之一就是顺序执行一系列非异步（同步）的任务。这些任务在同一个线程中执行，通常会阻塞主程序的运行，直到任务完成。

示例代码：

import timedef blocking_task(id):print(f"Blocking task {id} started")time.sleep(2)  # 模拟一个阻塞操作print(f"Blocking task {id} finished")# 顺序执行多个同步任务
def main():for i in range(3):blocking_task(i)main()

解释:

在此代码中，每个任务按顺序执行，time.sleep() 会阻塞当前线程，直到所有任务结束。这种顺序执行的方式虽然简单直接，但效率较低，尤其当任务涉及 I/O 操作时，会浪费大量时间。

3 顺序执行异步任务 ⏳

如果我们希望提高任务的执行效率，可以考虑使用异步任务。异步任务不会阻塞主线程，而是会等待特定的事件（例如 I/O 操作的完成），然后继续执行。

示例代码：

import asyncioasync def async_task(id):print(f"Async task {id} started")await asyncio.sleep(2)  # 模拟异步操作print(f"Async task {id} finished")# 顺序执行多个异步任务
async def main():for i in range(3):await async_task(i)asyncio.run(main())

解释：

通过使用 async def 定义异步函数，await 关键字用于暂停任务的执行并等待异步操作完成。虽然这些任务是异步的，但由于我们使用了 await，它们仍然是顺序执行的。

3 并行执行异步任务 🚀

在某些情况下，我们可能希望异步任务能够并行执行，而不是一个接一个地等待。此时可以使用 asyncio.gather()，它允许我们并行运行多个异步任务，从而提高程序效率。

示意图：
在这里插入图片描述

示例代码：

import asyncioasync def async_task(id):print(f"Async task {id} started")await asyncio.sleep(2)print(f"Async task {id} finished")# 并行执行多个异步任务
async def main():tasks = [async_task(i) for i in range(3)]await asyncio.gather(*tasks)asyncio.run(main())

解释：

asyncio.gather() 会并行执行多个异步任务，而不是按顺序等待。任务在后台同时运行，极大提高了效率，尤其是当任务需要等待 I/O 时（例如网络请求、文件操作等）。

❗ 注意：

tasks = [async_task(i) for i in range(3)] 这个时候不会执行async_task函数，只是创建协程对象，还没真正的启动。

当你调用 async_task(i) 时，它不会立即执行，而是返回一个协程对象（coroutine object），这个对象代表一个等待执行的异步任务。
只有当你 await 这个协程对象或者将它传递给 asyncio.gather()、asyncio.create_task()、asyncio.run() 等函数时，协程才会开始执行。

4 并行执行非异步任务（阻塞任务） 🛠️

如果你有一些外部库提供的阻塞任务（如文件读写、网络操作等），这些任务无法直接变为异步函数。为了与异步任务并行执行这些阻塞任务，asyncio.run_in_executor() 是你的好帮手。

示意图：
在这里插入图片描述

示例代码：使用线程池并行执行同步任务

import asyncio
import time
from concurrent.futures import ThreadPoolExecutordef blocking_task(id):print(f"Blocking task {id} started")time.sleep(2)print(f"Blocking task {id} finished")async def main():with ThreadPoolExecutor() as pool:tasks = [asyncio.get_event_loop().run_in_executor(pool, blocking_task, i)for i in range(3)]await asyncio.gather(*tasks)asyncio.run(main())

解释：

run_in_executor() 将阻塞的任务交给线程池（或进程池）执行，而不会阻塞主事件循环。这使得我们可以同时处理异步任务和阻塞任务。

使用进程池还是线程池？ 🤔

线程池（ThreadPoolExecutor）：适用于 I/O 密集型任务，如文件操作或网络请求。这类任务通常会等待外部事件完成，因此不需要消耗大量 CPU 资源。
进程池（ProcessPoolExecutor）：适合 CPU 密集型任务，如数据处理和计算。使用进程池可以充分利用多核 CPU，提升性能。

示例代码：使用进程池

from concurrent.futures import ProcessPoolExecutor
import asynciodef cpu_intensive_task(id):print(f"CPU task {id} started")result = sum(i*i for i in range(10**6))  # 模拟CPU密集任务print(f"CPU task {id} finished with result: {result}")async def main():with ProcessPoolExecutor() as pool:tasks = [asyncio.get_event_loop().run_in_executor(pool, cpu_intensive_task, i)for i in range(3)]await asyncio.gather(*tasks)asyncio.run(main())