Python 异步编程与协程实战指南

任务调度与管理

在实际开发中，我们需要对任务进行精细控制，比如取消超时任务或动态创建。

创建与取消任务

使用 asyncio.create_task() 将协程包装为 Task 对象，便于后续管理。

import asyncio

async def task_func():
    try:
        print('Task started')
        await asyncio.sleep(5)
        print('Task finished')
    except asyncio.CancelledError:
        print('Task was cancelled')

async def main():
    t = asyncio.create_task(task_func())
    await asyncio.sleep(1)
    t.cancel()  # 请求取消任务
    try:
        await t
    except asyncio.CancelledError:
        print('Main: Task cancelled successfully')

if __name__ == '__main__':
    asyncio.run(main())

设置超时保护

防止某个任务卡死影响整体流程，可以使用 asyncio.wait_for()。

import asyncio

async def slow_task():
    await asyncio.sleep(10)
    return 'Done'

async def main():
    try:
        # 限制执行时间为 2 秒
        result = await asyncio.wait_for(slow_task(), timeout=2)
        print(result)
    except asyncio.TimeoutError:
        print('Task timed out')

if __name__ == '__main__':
    asyncio.run(main())

事件循环机制

事件循环是异步引擎的心脏，负责轮询就绪的任务并分发执行。虽然现代 Python 推荐使用 asyncio.run() 自动管理循环，但在某些底层场景下，手动获取循环仍有必要。

import asyncio

loop = asyncio.get_event_loop()

async def worker():
    print('Worker running')
    await asyncio.sleep(1)

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

aiohttp 实战

aiohttp 是 Python 中最流行的异步 HTTP 客户端和服务端库。安装只需一行命令：pip install aiohttp。

异步 HTTP 客户端

下面是一个并发抓取多个网页的示例，对比同步方式能节省大量时间。

import aiohttp
import asyncio
import time

async def fetch(session, url):
    start_time = time.time()
    async with session.get(url) as response:
        text = await response.text()
        elapsed = time.time() - start_time
        return url, len(text), elapsed

async def main():
    urls = [
        'https://www.example.com',
        'https://www.google.com',
        'https://www.github.com',
        'https://www.python.org'
    ]
    
    async with aiohttp.ClientSession() as session:
        tasks = [fetch(session, url) for url in urls]
        results = await asyncio.gather(*tasks)
        
        for url, length, elapsed in results:
            print(f'URL: {url}, Size: {length}, Time: {elapsed:.2f}s')

if __name__ == '__main__':
    start = time.time()
    asyncio.run(main())
    print(f'Total Time: {time.time() - start:.2f}s')

构建异步 Web 服务器

利用 aiohttp.web 模块可以快速搭建高性能 API 服务。

from aiohttp import web
import asyncio

async def handle_root(request):
    return web.Response(text='Hello, Async World!')

async def handle_api(request):
    data = {'status': 'ok', 'message': 'Success'}
    return web.json_response(data)

async def create_app():
    app = web.Application()
    app.router.add_get('/', handle_root)
    app.router.add_get('/api', handle_api)
    return app

if __name__ == '__main__':
    # 启动服务器监听 8080 端口
    asyncio.run(web.run_app(create_app(), host='localhost', port=8080))

总结

掌握异步编程的关键在于理解事件循环如何调度协程。通过 asyncio 和 aiohttp，我们可以轻松构建高并发的网络应用。建议在实际项目中多尝试混合使用同步与异步组件，并注意避免在协程中调用阻塞式 IO，这样才能真正发挥异步的优势。