Python 异步 I/O 学习的 asyncio 库

xiaoshi 05-30 17 抢沙发

默认

摘要： ...

Python异步编程利器：深入理解asyncio库的核心用法

为什么需要异步I/O？

在传统的同步编程中，当一个程序执行I/O操作（如网络请求、文件读写）时，整个线程会被阻塞，直到操作完成。这种方式在并发量大的场景下效率极低，因为大部分时间都在等待I/O完成，而不是真正执行计算任务。

异步I/O通过事件循环机制解决了这个问题。当遇到I/O操作时，程序不会阻塞等待，而是继续执行其他任务，等I/O完成后通过回调或协程机制处理结果。这种非阻塞方式极大提高了程序的吞吐量，特别适合网络应用、爬虫等I/O密集型场景。

asyncio库的基本概念

asyncio是Python标准库中用于编写单线程并发代码的模块，使用async/await语法。它的核心是事件循环(event loop)，负责调度和执行异步任务。

理解asyncio需要掌握几个关键概念：

协程(Coroutine)：使用async def定义的函数，是异步编程的基本单元
任务(Task)：对协程的进一步封装，表示一个正在运行或将要运行的协程
Future：表示异步操作的最终结果，是Task的基类
事件循环(Event Loop)：核心调度器，负责协调所有异步任务的执行

快速上手asyncio

让我们从一个简单的例子开始：

import asyncio

async def say_hello():
    print("Hello")
    await asyncio.sleep(1)
    print("World")

async def main():
    await asyncio.gather(say_hello(), say_hello(), say_hello())

asyncio.run(main())

这段代码会同时启动三个say_hello协程，虽然每个协程都有1秒的等待时间，但由于是异步执行，总运行时间接近1秒而非3秒。

核心API详解

1. 事件循环管理

asyncio提供了多种管理事件循环的方式：

# 获取当前事件循环
loop = asyncio.get_event_loop()

# 运行协程直到完成
loop.run_until_complete(main())

# 关闭事件循环
loop.close()

# Python 3.7+ 更简单的写法
asyncio.run(main())

2. 任务创建与控制

async def my_task():
    print("Task started")
    await asyncio.sleep(2)
    print("Task completed")

async def main():
    # 创建任务
    task1 = asyncio.create_task(my_task())
    task2 = asyncio.create_task(my_task())

    # 等待任务完成
    await task1
    await task2

3. 超时控制

async def long_running_task():
    await asyncio.sleep(10)
    return "Done"

async def main():
    try:
        # 设置5秒超时
        result = await asyncio.wait_for(long_running_task(), timeout=5.0)
    except asyncio.TimeoutError:
        print("Task timed out")

高级用法与最佳实践

1. 并发执行多个协程

asyncio.gather()可以并发运行多个协程：

async def fetch_data(url):
    # 模拟网络请求
    await asyncio.sleep(1)
    return f"Data from {url}"

async def main():
    results = await asyncio.gather(
        fetch_data("url1"),
        fetch_data("url2"),
        fetch_data("url3")
    )
    print(results)

2. 使用队列实现生产者消费者模式

async def producer(queue):
    for i in range(5):
        await queue.put(i)
        await asyncio.sleep(0.5)

async def consumer(queue):
    while True:
        item = await queue.get()
        print(f"Consumed {item}")
        queue.task_done()

async def main():
    queue = asyncio.Queue()
    await asyncio.gather(
        producer(queue),
        consumer(queue)
    )

3. 协程与线程池结合

对于CPU密集型任务，可以结合线程池使用：

import concurrent.futures

def cpu_bound_task(x):
    # CPU密集型计算
    return x * x

async def main():
    loop = asyncio.get_running_loop()

    with concurrent.futures.ThreadPoolExecutor() as pool:
        result = await loop.run_in_executor(pool, cpu_bound_task, 10)
        print(result)

常见问题与解决方案

1. 如何避免阻塞事件循环？

避免在协程中执行同步I/O操作或长时间运行的CPU密集型任务。如果必须执行这类操作，可以使用run_in_executor将其放到线程池中执行。

2. 协程中如何处理异常？

协程中的异常处理与普通函数类似：

async def risky_task():
    try:
        await asyncio.sleep(1)
        raise ValueError("Something went wrong")
    except ValueError as e:
        print(f"Caught exception: {e}")

3. 如何取消正在运行的任务？

async def long_task():
    try:
        await asyncio.sleep(3600)
    except asyncio.CancelledError:
        print("Task was cancelled")
        raise

async def main():
    task = asyncio.create_task(long_task())
    await asyncio.sleep(1)
    task.cancel()
    try:
        await task
    except asyncio.CancelledError:
        print("Main caught cancellation")

性能优化技巧

合理设置并发限制：使用信号量(Semaphore)控制最大并发数

async def limited_task(semaphore, url):
    async with semaphore:
        return await fetch_data(url)

async def main():
    semaphore = asyncio.Semaphore(10)  # 最大并发10
    tasks = [limited_task(semaphore, url) for url in urls]
    await asyncio.gather(*tasks)

复用连接：对于网络请求，复用连接池比每次创建新连接更高效
避免过早await：将多个I/O操作批量处理后再await，减少上下文切换

实际应用案例

1. 高性能网络爬虫

async def crawl(url):
    print(f"Crawling {url}")
    await asyncio.sleep(1)  # 模拟网络请求
    return f"Content from {url}"

async def crawl_all(urls):
    tasks = [crawl(url) for url in urls]
    return await asyncio.gather(*tasks)

urls = [f"https://example.com/page{i}" for i in range(10)]
results = asyncio.run(crawl_all(urls))

2. WebSocket客户端

import websockets

async def websocket_client():
    async with websockets.connect("ws://example.com") as websocket:
        await websocket.send("Hello")
        response = await websocket.recv()
        print(response)

asyncio.run(websocket_client())