异步深水区：异常传播、超时取消与阻塞陷阱

import asyncio

async def fetch(name, delay, fail=False):
    await asyncio.sleep(delay)
    if fail:
        raise ConnectionError(f"{name} 连接失败")
    print(f"{name} 完成")
    return name

async def main():
    try:
        await asyncio.gather(
            fetch("页面A", 2),
            fetch("页面B", 1, fail=True),
        )
    except ConnectionError as e:
        print(f"gather 在第 1 秒抛出: {e}")
    print("—— 此时页面A 还在跑，再等它 1.5 秒 ——")
    await asyncio.sleep(1.5)
    print("main 结束，页面A 早已完成（gather 没有取消兄弟任务）")

asyncio.run(main())

gather 在第 1 秒抛出: 页面B 连接失败
—— 此时页面A 还在跑，再等它 1.5 秒 ——
页面A 完成
main 结束，页面A 早已完成（gather 没有取消兄弟任务）

    results = await asyncio.gather(
        fetch("页面A", 0.2),
        fetch("页面B", 0.1, fail=True),
        fetch("页面C", 0.3),
        return_exceptions=True,
    )
    for r in results:
        print(repr(r))

'页面A'
ConnectionError('页面B 连接失败')
'页面C'

import asyncio

async def fetch(url: str) -> dict:
    await asyncio.sleep(1)               # 模拟网络请求
    return {"url": url, "status": 200}

async def fetch_all() -> None:
    async with asyncio.TaskGroup() as tg:
        t1 = tg.create_task(fetch("https://api.example.com/users"))
        t2 = tg.create_task(fetch("https://api.example.com/posts"))
        t3 = tg.create_task(fetch("https://api.example.com/comments"))
    # 走到这里说明三个任务全部成功——失败走不到这里
    results = [t.result() for t in (t1, t2, t3)]
    print(f"获取 {len(results)} 个资源")

asyncio.run(fetch_all())

import asyncio

async def fetch_slow():
    try:
        print("开始下载（预计 10 秒）")
        await asyncio.sleep(10)
        return "内容"
    except asyncio.CancelledError:
        print("fetch_slow 收到取消：超时把我也带走了")
        raise                                  # 必须原样抛回

async def main():
    try:
        result = await asyncio.wait_for(fetch_slow(), timeout=2)
        print(result)
    except TimeoutError as e:
        print(f"main 捕获 TimeoutError: {e!r}")

asyncio.run(main())

开始下载（预计 10 秒）
fetch_slow 收到取消：超时把我也带走了
main 捕获 TimeoutError: TimeoutError()

async def resilient_fetch() -> None:
    try:
        async with asyncio.timeout(3.0):
            async with asyncio.TaskGroup() as tg:
                tg.create_task(fetch("https://api1.example.com"))
                tg.create_task(fetch("https://api2.example.com"))
    except TimeoutError:
        print("限时已到，组内任务已被全部取消")

async def main():
    task = asyncio.create_task(fetch_slow())
    await asyncio.sleep(1)
    task.cancel()
    try:
        await task
    except asyncio.CancelledError:
        print("main 确认：task 已被取消")

asyncio.run(main())

开始下载（预计 10 秒）
fetch_slow 收到取消：超时把我也带走了
main 确认：task 已被取消

import asyncio
from contextlib import asynccontextmanager

@asynccontextmanager
async def db_connection(name):
    print(f"[{name}] 连接数据库")
    await asyncio.sleep(0.1)              # 模拟异步建连
    try:
        yield "conn"                      # yield 前是进入，之后是退出
    finally:
        print(f"[{name}] 关闭连接")
        await asyncio.sleep(0.1)          # 关闭也是 IO，也需要 await

async def query(conn, sql):
    await asyncio.sleep(0.2)
    return f"{sql} 的结果"

async def main():
    async with db_connection("订单库") as conn:
        result = await query(conn, "SELECT count(*) FROM orders")
        print(result)

asyncio.run(main())

[订单库] 连接数据库
SELECT count(*) FROM orders 的结果
[订单库] 关闭连接

async def stream_events():
    """模拟事件流：每 0.5 秒吐一条"""
    for i in range(5):
        await asyncio.sleep(0.5)
        yield {"event_id": i, "data": f"事件 {i}"}

async def consume_events() -> None:
    async for event in stream_events():
        print(f"收到: {event['event_id']}")

asyncio.run(consume_events())

async def work_bad(name):
    print(f"{name} 开始")
    time.sleep(1)                 # 阻塞整个事件循环 1 秒
    print(f"{name} 结束")

async def main_bad():
    start = time.perf_counter()
    await asyncio.gather(work_bad("任务A"), work_bad("任务B"), work_bad("任务C"))
    print(f"time.sleep 版总耗时 {time.perf_counter() - start:.1f} 秒")

asyncio.run(main_bad())

任务A 开始
任务A 结束
任务B 开始
任务B 结束
任务C 开始
任务C 结束
time.sleep 版总耗时 3.0 秒

任务A 开始
任务B 开始
任务C 开始
任务A 结束
任务B 结束
任务C 结束
asyncio.sleep 版总耗时 1.0 秒

result = await asyncio.to_thread(time.sleep, 1)   # 阻塞发生在线程里，循环照常转

import asyncio
import aiofiles

# 路线一：aiofiles——语法与同步 open 几乎一致，await 替换阻塞点
async def read_config(path: str) -> str:
    async with aiofiles.open(path, "r", encoding="utf-8") as f:
        content = await f.read()
    return content

async def write_report(path: str, content: str) -> None:
    async with aiofiles.open(path, "w", encoding="utf-8") as f:
        await f.write(content)

# 路线二：to_thread 包同步读——不引第三方库，大块读写一次进线程
async def read_config_stdlib(path: str) -> str:
    return await asyncio.to_thread(lambda: open(path, "r", encoding="utf-8").read())

async def main():
    asyncio.create_task(failing_task())
    await asyncio.sleep(1)

asyncio.run(main())

Task exception was never retrieved
future: <Task finished ... exception=ConnectionError('后台任务炸了')>
Traceback (most recent call last):
  File "exp_orphan.py", line 5, in failing_task
    raise ConnectionError("后台任务炸了")
ConnectionError: 后台任务炸了

import asyncio

async def worker(name, delay, fail=False):
    await asyncio.sleep(delay)
    if fail:
        raise ValueError(f"{name} 失败")
    return name

async def main():
    results = await asyncio.gather(
        worker("A", 0.2),
        worker("B", 0.1, fail=True),
        worker("C", 0.3),
        return_exceptions=True,
    )
    for r in results:
        print(repr(r))

asyncio.run(main())

async def main():
    start = time.perf_counter()
    try:
        await run_with_cleanup(slow_download(), timeout=1)
        assert False, "应当超时"
    except TimeoutError:
        elapsed = time.perf_counter() - start
        assert 0.9 < elapsed < 2, f"超时未生效: {elapsed:.2f} 秒"

asyncio.run(main())