Python 异步编程模式：从协程到事件循环深度剖析

import asyncio
import aiohttp
from typing import List

async def fetch_url(session: aiohttp.ClientSession, url: str) -> dict:
    """异步获取 URL 内容"""
    async with session.get(url) as response:
        return await response.json()

async def fetch_all(urls: List[str]) -> List[dict]:
    """并发获取所有 URL"""
    async with aiohttp.ClientSession() as session:
        tasks = [fetch_url(session, url) for url in urls]
        return await asyncio.gather(*tasks, return_exceptions=True)

async def main():
    urls = [
        "https://api.example.com/users",
        "https://api.example.com/posts",
        "https://api.example.com/comments",
    ]
    results = await fetch_all(urls)
    for i, result in enumerate(results):
        print(f"请求 {i}: {'成功' if not isinstance(result, Exception) else '失败'}")

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

#!/bin/bash
# 异步程序运行脚本
python3 -m pip install aiohttp  # 安装依赖
python3 async_fetch.py          # 运行异步程序