Python：asyncio —— 一个事件循环，和它的染色问题

事件循环：把它拆开，只有三样东西

剥掉所有的糖，asyncio 的事件循环就是一个 while 循环 + 两个队列：

def run_forever(self):                 # 这就是 asyncio 的心脏(极简版)
    while True:
        # ① 有没有到期的定时器? 把它们的回调塞进 ready
        now = time.monotonic()
        while self._timers and self._timers[0].when <= now:
            self._ready.append(heapq.heappop(self._timers).callback)

        # ② 等 IO —— 这就是第 5 章那个 epoll_wait!
        #    如果 ready 队列不空,timeout=0(不等,马上回来干活)
        #    如果 ready 队列空了,就睡到最早的定时器到期为止(0 CPU)
        timeout = 0 if self._ready else self._time_to_next_timer()
        events = self._selector.select(timeout)      # ← select/epoll/kqueue
        for fd, callback in events:
            self._ready.append(callback)             # IO 好了 → 塞进 ready

        # ③ 把 ready 队列里的回调全跑一遍
        for callback in self._ready.pop_all():
            callback()                               # ← 这里绝对不能阻塞!!!

下面这个 demo 是一个真的（极小的）事件循环。先按「单步」走一遍正常情况，看 ready 队列和定时器堆怎么动、循环什么时候去睡。然后切到第二个 tab —— 有人偷偷用了 time.sleep。

协程对象：它其实是个生成器

Python 的协程不是魔法，它是生成器（generator）的一个变种。这个血缘关系解释了它的一切行为：

async def foo():
    print("A")
    await asyncio.sleep(1)
    print("B")

c = foo()          # ← 注意!这里什么也没执行!
print(type(c))     #   ← 你只是造了一个对象

# 协程对象是「惰性」的 —— 你必须把它交给事件循环,它才会跑:
asyncio.run(foo())              # 或者
task = asyncio.create_task(foo())   # 或者
await foo()

async def main():
    save_to_db(data)        # ❌ 忘了 await!
    return "ok"             # 函数返回了，而 save_to_db 一行都没执行过

# Python 会好心地警告你:
#   RuntimeWarning: coroutine 'save_to_db' was never awaited
#
# 但它只是个 warning。在生产日志里,它会淹没在一万行 INFO 里。
# 而你的数据,静静地没有被保存。

那个会毁掉一切的东西：阻塞调用

上面的 demo 已经演过了。这里列一份通缉名单：

import asyncio

async def handler(req):
    # ✅ 把阻塞调用丢进线程池,事件循环继续转
    result = await asyncio.to_thread(legacy_blocking_sdk.query, req.id)
    #        ^^^^^^^^^^^^^^^^^^^^^^ Python 3.9+
    return result

# 它做的事:
#   1. 把 legacy_blocking_sdk.query 提交给一个 ThreadPoolExecutor
#   2. 当前协程挂起(线程还给循环)
#   3. 线程池里的那个线程去阻塞(它阻塞就阻塞吧,反正 GIL 会被 IO 释放)
#   4. 干完了,唤醒协程

怎么并发：三个 API，和它们的取舍

① gather —— 老 API，有个坑

results = await asyncio.gather(
    fetch(url1),
    fetch(url2),
    fetch(url3),
)   # 三个请求并发,全部完成后返回

# ⚠️ 坑:如果 url2 抛异常了呢?
#   默认行为:gather 立刻把异常抛给你 ——
#   但 url1 和 url3 的 task 还在后台跑!没人取消它们,没人等它们。
#   这就是「孤儿 task」。

# 半个补丁:
results = await asyncio.gather(*tasks, return_exceptions=True)
# 现在异常变成了返回值里的一项,不会中断其他 task。
# 但你得自己逐个检查每个 result 是不是 Exception。

② TaskGroup（3.11+）—— 用这个

async def main():
    async with asyncio.TaskGroup() as tg:      # Python 3.11+
        t1 = tg.create_task(fetch(url1))
        t2 = tg.create_task(fetch(url2))
        t3 = tg.create_task(fetch(url3))
    # ↑ 退出 with 块时,保证:所有 task 要么全部完成,要么全部被取消。
    #   没有孤儿。

    print(t1.result(), t2.result(), t3.result())

# 如果 fetch(url2) 抛了异常:
#   1. TaskGroup 立刻取消 t1 和 t3
#   2. 等它们真的取消完
#   3. 把所有异常打包成 ExceptionGroup 抛出来

# 配套的 except* 语法(也是 3.11+):
try:
    async with asyncio.TaskGroup() as tg:
        ...
except* ConnectionError as eg:      # 注意那个星号
    for e in eg.exceptions:
        log.warning("网络挂了: %s", e)
except* ValueError as eg:
    for e in eg.exceptions:
        log.error("数据错了: %s", e)
# 因为可能有多个 task 同时失败,异常是「一组」而不是「一个」。

③ create_task 裸用 —— 危险

# ❌ 这是 Python 版的 `go f()` —— 一个没人管的孤儿
asyncio.create_task(background_job())
# 函数返回了。这个 task 还在跑。谁等它?谁取消它?谁看它的异常?

# ⚠️ 而且还有一个真实的坑:
#   事件循环只持有 task 的「弱引用」!
#   如果你不保存返回值,task 可能被 GC 掉,任务凭空消失。
#   (官方文档明确警告了这一点。)

# 如果非要用,至少:
_background_tasks = set()

def spawn(coro):
    t = asyncio.create_task(coro)
    _background_tasks.add(t)                    # 保持强引用
    t.add_done_callback(_background_tasks.discard)
    t.add_done_callback(_log_exception)         # 否则异常会被静默吞掉!
    return t

ERROR:asyncio:Task exception was never retrieved
future: <Task finished coro=<background_job() done> exception=ValueError('boom')>

超时和取消

# ✅ 3.11+ 的正确姿势
async with asyncio.timeout(5):        # 5 秒超时
    await fetch(url)
# 超时时:向这个协程注入一个 CancelledError

# 更细的控制
async with asyncio.timeout(5) as cm:
    cm.reschedule(loop.time() + 10)   # 中途改主意,延长到 10 秒
    await fetch(url)

# 老 API(还能用,但不如上面的组合性好)
await asyncio.wait_for(fetch(url), timeout=5)

# ❌ 灾难
async def worker():
    try:
        await do_work()
    except Exception:          # CancelledError 在 3.8+ 继承自 BaseException,
        log.error("出错了")     # 所以这里【抓不到】它 —— 这是好事!
        # ↑ 3.8 之前它继承 Exception,这里会把取消吞掉,导致取消失效

# ❌ 但这个还是灾难(3.8+ 也一样)
async def worker():
    try:
        await do_work()
    except BaseException:      # ← 抓到了 CancelledError
        log.error("出错了")     # ← 然后【没有重新抛出】
        # 取消信号被你吞了。上层以为取消成功了,而这个 task 还在跑。

# ✅ 正确:清理,然后一定要重新抛出
async def worker():
    try:
        await do_work()
    except asyncio.CancelledError:
        await cleanup()        # 做你的清理
        raise                  # ← 必须!把取消传下去

诚实地评价 asyncio

下一章：换一门语言。Kotlin 的 suspend —— 那个「看起来像同步、跑起来非阻塞」的魔法，编译器到底做了什么。（剧透：一台状态机。）