Python 异步编程：asyncio 深入理解与实战

引言

Python 的 asyncio 库为编写并发异步代码提供了强大的支持。然而，很多开发者对其内部机制理解不深，导致难以编写出高效的异步程序。本文将深入探讨 asyncio 的核心组件，并通过实战案例帮助你掌握异步编程的最佳实践。

1. 事件循环与协程

1.1 事件循环

事件循环是 asyncio 的核心，它负责调度和执行协程。在 Python 3.7+ 中，推荐使用 asyncio.run() 来启动事件循环。


import asyncio

async def main():
    print('Hello')
    await asyncio.sleep(1)
    print('World')

asyncio.run(main())

1.2 协程

协程是通过 async def 定义的函数，调用它会返回一个协程对象。协程对象需要被事件循环调度才能执行。


async def foo():
    return 42

coro = foo()  # 协程对象
result = await coro  # 等待结果

2. 任务 (Task)

任务用于并发执行协程。通过 asyncio.create_task() 可以将协程包装成任务，并立即开始调度。


async def say_after(delay, msg):
    await asyncio.sleep(delay)
    print(msg)

async def main():
    task1 = asyncio.create_task(say_after(1, 'Hello'))
    task2 = asyncio.create_task(say_after(2, 'World'))
    await task1
    await task2

asyncio.run(main())

注意：create_task 会立即将协程注册到事件循环，但只有事件循环运行时才会执行。

3. Future 对象

Future 是底层对象，表示异步操作的最终结果。Task 是 Future 的子类。通常我们不需要直接操作 Future，但理解它有助于深入理解 asyncio。


async def set_after(fut, delay, value):
    await asyncio.sleep(delay)
    fut.set_result(value)

async def main():
    loop = asyncio.get_running_loop()
    fut = loop.create_future()
    asyncio.create_task(set_after(fut, 1, '... done'))
    result = await fut
    print(result)

asyncio.run(main())

4. 并发控制：gather 与 wait

4.1 asyncio.gather

gather 用于并发运行多个协程，并等待所有完成。


async def fetch_data(delay):
    await asyncio.sleep(delay)
    return f'Data after {delay}s'

async def main():
    results = await asyncio.gather(
        fetch_data(1),
        fetch_data(2),
        fetch_data(3)
    )
    print(results)

asyncio.run(main())

4.2 asyncio.wait

wait 提供更细粒度的控制，可以等待第一个完成、所有完成等。


async def main():
    tasks = [asyncio.create_task(fetch_data(i)) for i in [1, 2, 3]]
    done, pending = await asyncio.wait(tasks, return_when=asyncio.FIRST_COMPLETED)
    for task in done:
        print(task.result())
    for task in pending:
        task.cancel()

asyncio.run(main())

5. 避免常见陷阱

5.1 不要在协程中使用 time.sleep

time.sleep 会阻塞整个线程，应使用 asyncio.sleep。


# 错误
async def bad():
    import time
    time.sleep(1)  # 阻塞事件循环

# 正确
async def good():
    await asyncio.sleep(1)

5.2 避免协程中执行 CPU 密集型任务

CPU 密集型任务会阻塞事件循环，应使用 asyncio.to_thread 或 concurrent.futures 将其转移到线程池。


import asyncio

def cpu_bound():
    count = 0
    for i in range(10**7):
        count += i
    return count

async def main():
    result = await asyncio.to_thread(cpu_bound)
    print(result)

asyncio.run(main())

5.3 正确使用锁

asyncio 提供了协程安全的锁 asyncio.Lock。


async def worker(lock, name):
    async with lock:
        print(f'{name} acquired lock')
        await asyncio.sleep(1)
        print(f'{name} released lock')

async def main():
    lock = asyncio.Lock()
    await asyncio.gather(
        worker(lock, 'A'),
        worker(lock, 'B')
    )

asyncio.run(main())

6. 实战：异步 Web 爬虫

结合 aiohttp 实现一个简单的异步爬虫。


import asyncio
import aiohttp

async def fetch(session, url):
    async with session.get(url) as response:
        return await response.text()

async def main():
    urls = [
        'https://example.com',
        'https://httpbin.org/get',
        'https://jsonplaceholder.typicode.com/posts/1'
    ]
    async with aiohttp.ClientSession() as session:
        tasks = [asyncio.create_task(fetch(session, url)) for url in urls]
        results = await asyncio.gather(*tasks)
        for result in results:
            print(len(result))

asyncio.run(main())

注意：使用 aiohttp 时，需要安装 aiohttp 库。

总结

本文深入介绍了 asyncio 的核心概念，包括事件循环、协程、任务和 Future，并通过实战案例展示了如何编写高效的异步代码。记住：避免阻塞事件循环，合理使用并发控制，正确处理 IO 和 CPU 密集型任务。

引言