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

三、任务调度

3.1 创建任务

import asyncio

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

async def main():
    task1 = asyncio.create_task(hello())
    task2 = asyncio.create_task(hello())
    await task1
    await task2

asyncio.run(main())

3.2 任务取消

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

3.3 任务超时

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

四、事件循环

4.1 概述

事件循环是异步编程核心组件，负责调度任务执行，从队列取出任务直到为空。

4.2 获取与运行

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

五、asyncio 库

5.1 常用函数

• asyncio.run()：运行协程。
• asyncio.create_task()：创建任务。
• asyncio.gather()：收集多个协程结果。
• asyncio.wait_for()：设置超时等待。
• asyncio.sleep()：暂停协程。

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

六、aiohttp 库

6.1 安装

pip install aiohttp

6.2 发送 HTTP 请求

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

6.3 POST 与 JSON 请求

import aiohttp
import asyncio
import json

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

七、实战案例

7.1 异步 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.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']
    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}秒')

7.2 异步 Web 服务器

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 create_app():
    app = web.Application()
    app.add_routes([
        web.get('/', handle_root),
        web.get('/api', handle_api)
    ])
    return app

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

八、总结

本文介绍了 Python 异步编程的核心概念，包括协程、任务调度、事件循环及 asyncio 和 aiohttp 库的使用。通过实战案例展示了异步 HTTP 客户端和 Web 服务器的开发方法，帮助开发者掌握高并发网络应用技巧。