Python中asyncio的多种用法举例(异步同步)_Python

1 引言

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

2 顺序执行非异步任务

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

示例代码：

import time

def 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 asyncio

async 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 asyncio

async 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 threadpoolexecutor

def 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 asyncio

def 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())