Python 异步编程简明指南：协程、任务与 asyncio

取消一个任务：

import asyncio

async def hello():
    try:
        print('Hello, World!')
        await asyncio.sleep(1)
        print('Hello again!')
    except asyncio.CancelledError:
        print('Task cancelled!')

async def main():
    task = asyncio.create_task(hello())
    await asyncio.sleep(0.5)
    task.cancel()
    try:
        await task
    except asyncio.CancelledError:
        print('Main: Task cancelled!')

asyncio.run(main())

还可以用 asyncio.wait_for 给任务加上超时：

import asyncio

async def hello():
    print('Hello, World!')
    await asyncio.sleep(2)
    print('Hello again!')

async def main():
    try:
        await asyncio.wait_for(hello(), timeout=1)
    except asyncio.TimeoutError:
        print('Task timed out!')

asyncio.run(main())

事件循环：幕后调度器

事件循环负责调度所有协程和任务。通常我们用 asyncio.run 就够了，它会新建并运行一个事件循环。但也可以直接拿到事件循环来执行协程：

import asyncio

loop = asyncio.get_event_loop()

async def hello():
    print('Hello, World!')
    await asyncio.sleep(1)
    print('Hello again!')

loop.run_until_complete(hello())

或者并发执行多个：

import asyncio

loop = asyncio.get_event_loop()

async def hello():
    print('Hello, World!')
    await asyncio.sleep(1)
    print('Hello again!')

loop.run_until_complete(asyncio.gather(hello(), hello(), hello()))

不过在新代码里，更推荐 asyncio.run。

asyncio 核心函数速览

asyncio 提供了几个常用函数：

• asyncio.run() – 运行一个协程，同时启动事件循环。
• asyncio.create_task() – 把协程包装成任务。
• asyncio.gather() – 并发执行多个协程并收集结果。
• asyncio.wait_for() – 等待协程，可设置超时。
• asyncio.sleep() – 在协程里暂停，不阻塞循环。

还有像 asyncio.Future 这样的底层对象，用来桥接同步代码。下面是一个使用 Future 的例子：

import asyncio

async def hello():
    print('Hello, World!')
    await asyncio.sleep(1)
    print('Hello again!')

async def main():
    future = asyncio.Future()
    task = asyncio.create_task(hello())
    task.add_done_callback(lambda t: future.set_result(t.result()))
    await future

asyncio.run(main())

异步 HTTP 请求：用 aiohttp

aiohttp 是异步 HTTP 客户端和服务器库。先安装：

pip install aiohttp

发送 GET 请求：

import aiohttp
import asyncio

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

async def main():
    async with aiohttp.ClientSession() as session:
        html = await fetch(session, 'https://www.example.com')
        print(html)

asyncio.run(main())

POST 表单数据：

import aiohttp
import asyncio

async def post_data(session, url, data):
    async with session.post(url, data=data) as response:
        return await response.text()

async def main():
    async with aiohttp.ClientSession() as session:
        data = {'name': '张三', 'age': 25}
        response = await post_data(session, 'https://httpbin.org/post', data)
        print(response)

asyncio.run(main())

发送 JSON 数据：

import aiohttp
import asyncio

async def post_json(session, url, data):
    async with session.post(url, json=data) as response:
        return await response.text()

async def main():
    async with aiohttp.ClientSession() as session:
        data = {'name': '张三', 'age': 25}
        response = await post_json(session, 'https://httpbin.org/post', data)
        print(response)

asyncio.run(main())

实战：并发 HTTP 客户端

很多时候你需要并发请求多个 URL。下面这个脚本会同时请求五个网站，输出每个请求的响应长度和耗时：

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.time() - start_time
        return url, len(text), elapsed_time

async def main():
    urls = ['https://www.example.com', 'https://www.google.com', 'https://www.github.com', 'https://www.python.org', 'https://www.djangoproject.com']
    async with aiohttp.ClientSession() as session:
        tasks = [asyncio.create_task(fetch(session, url)) for url in urls]
        results = await asyncio.gather(*tasks)
        for url, length, elapsed_time in results:
            print(f'URL: {url}, 响应长度：{length}, 耗时：{elapsed_time:.2f}秒')

if __name__ == '__main__':
    start_time = time.time()
    asyncio.run(main())
    elapsed_time = time.time() - start_time
    print(f'总耗时：{elapsed_time:.2f}秒')

这个脚本里，每个请求独立计时，但你也会注意到总耗时远小于所有请求时长的累加，这正是并发的效果。

实战：用 aiohttp 写个 Web 服务器

aiohttp 也能写服务器。下面这个例子提供了根路由、一个 JSON API 以及静态文件服务：

from aiohttp import web
import asyncio

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

async def handle_api(request):
    data = {'name': '张三', 'age': 25}
    return web.json_response(data)

async def handle_static(request):
    return web.FileResponse('static/index.html')

async def create_app():
    app = web.Application()
    app.add_routes([
        web.get('/', handle_root),
        web.get('/api', handle_api),
        web.get('/static/{name}', handle_static)
    ])
    return app

if __name__ == '__main__':
    asyncio.run(web.run_app(create_app(), host='localhost', port=8080))

运行后访问 http://localhost:8080 就能看到效果。

什么时候用异步

异步编程适合 I/O 密集型任务，比如发网络请求、读写文件、数据库查询。如果你的任务是 CPU 密集型的，asyncio 帮不上忙，反而可能降低性能。另外，异步代码需要你显式地使用 await，否则不会有并发效果；一些第三方库如果不支持异步，还得用 loop.run_in_executor 来执行同步代码。尽管有些学习成本，但对于高并发场景，asyncio 是一种很自然的选择。