CPython asyncio 开发实践完全指南：调试模式、多线程协作、阻塞代码治理与异步生成器陷阱

2026-09-06 18:39:40作者：余洋婵Anita

asyncio 是 CPython 内置的异步 I/O 框架，其编程范式与传统"顺序"编程存在本质差异，稍不注意就会踩进"协程未被等待""任务异常无人取回""异步生成器在错误时机被回收"等陷阱。本文以 CPython 仓库中的官方开发指南 Doc/library/asyncio-dev.rst 为核心骨架，逐条解读调试模式、并发与多线程边界、阻塞代码迁移、日志治理等常见问题，并结合 Lib/asyncio 目录下 base_events.py、coroutines.py、log.py 等源码实现，帮助你写出正确、高效、易于排查的异步程序。读完你将掌握：如何启用 asyncio 调试模式并利用其告警定位隐患、如何跨线程安全地与事件循环交互、何时必须将 CPU 密集代码交给执行器，以及三类异步生成器陷阱的成因与标准解法。

从认知差异说起：异步编程为何容易出错

"异步编程不同于经典的顺序编程"是理解本文一切内容的前提。在一个事件循环（event loop）内部，所有回调（callback）与 Task 都以协作式方式调度——一个 Task 运行期间同线程内不会并行执行其他 Task；只有当 Task 遇到 await 表达式并挂起时，事件循环才会切换去执行下一个 Task。这与多线程抢占式调度完全不同：

协程之间通过 await 主动让出控制权，而不是被操作系统时钟切片；
一旦某个环节不配合（例如直接调用了 CPU 密集函数或阻塞 I/O），整个事件循环里的所有任务都会跟着延迟；
因为切换点完全由代码决定，各种"没人调度它""没人处理它的异常""它被过早/过晚地回收"的问题就会以非直觉的方式暴露出来。

本页面向开发者列举的就是这些最常见的错误与陷阱，以及官方推荐的规避方式。下面逐一展开。

调试模式（Debug Mode）

默认情况下 asyncio 运行在 production mode（生产模式），该模式下为了性能几乎不做任何额外的运行时检查。为方便开发，asyncio 提供了 debug mode（调试模式）。

启用调试模式的四种途径

方式	说明
设置环境变量 `PYTHONASYNCIODEBUG=1`	进程级生效，影响该进程创建的所有事件循环
启用 Python 开发模式（Development Mode）	即 `python -X dev`，会顺带开启若干运行时检查
调用 `asyncio.run(..., debug=True)`	仅对本次运行生效，是最常用的临时开启方式
调用事件循环的 `loop.set_debug(True)`	可针对正在运行的事件循环动态开关

从源码可以看到这些途径的汇聚点：Lib/asyncio/coroutines.py 中 _is_debug_mode() 读取了环境变量与 sys.flags.dev_mode：

def _is_debug_mode():
    # See: https://docs.python.org/3/library/asyncio-dev.html#asyncio-debug-mode.
    return sys.flags.dev_mode or (not sys.flags.ignore_environment and
                                  bool(os.environ.get('PYTHONASYNCIODEBUG')))

而 Lib/asyncio/base_events.py 在 BaseEventLoop.__init__ 中通过 self.set_debug(coroutines._is_debug_mode()) 自动继承这一结果；asyncio.run(debug=True) 与 loop.set_debug(True) 则走 set_debug() 方法（base_events.py）动态覆写。

除开启调试模式外，官方还建议同时做两件事：

把 asyncio 日志器的级别调到 logging.DEBUG，例如在应用启动阶段执行：
```
logging.basicConfig(level=logging.DEBUG)
```
配置 warnings 模块展示 ResourceWarning 告警，一种方式是使用命令行选项 -W default 运行解释器，即 python -W default your_app.py。

调试模式下会触发哪些检查

开启调试模式后，asyncio 会带来如下额外行为：

非线程安全 API 的误用会直接抛异常：诸如 loop.call_soon()、loop.call_at() 等方法，一旦被从错误的线程调用，就会抛出异常而不是静默出错。原因可见 Lib/asyncio/base_events.py 的 _check_thread()：
```
def _check_thread(self):
    """Check that the current thread is the thread running the event loop.
    ...
    """
    if self._thread_id is None:
        return
    thread_id = threading.get_ident()
    if thread_id != self._thread_id:
        raise RuntimeError(
            "Non-thread-safe operation invoked on an event loop other "
            "than the current one")
```
call_soon() 内部在 self._debug 为真时会先调用 _check_thread()（见 base_events.py）。注意该检查只有在 debug 模式下才会执行——这正是生产模式"快但静默出错"、调试模式"慢但及早暴露问题"的典型取舍。
I/O selector 耗时过长会被记录日志：当一次 I/O 轮询（select/poll/epoll）耗时异常时，事件循环会打印相关日志，帮助发现可能导致事件循环停顿的异常 I/O。

执行时间超过 100 毫秒的回调会被记录：这条"慢回调"告警的阈值由 loop.slow_callback_duration 属性控制，单位为秒，默认为 0.1（100ms），在 base_events.py 初始化时设置：

# In debug mode, if the execution of a callback or a step of a task
# exceed this duration in seconds, the slow callback/task is logged.
self.slow_callback_duration = 0.1

其告警点位于事件循环单次迭代的核心 _run_once() 中（base_events.py）：

# This is the only place where callbacks are actually *called*.
ntodo = len(self._ready)
for i in range(ntodo):
    handle = self._ready.popleft()
    if handle._cancelled:
        continue
    if self._debug:
        try:
            self._current_handle = handle
            t0 = self.time()
            handle._run()
            dt = self.time() - t0
            if dt >= self.slow_callback_duration:
                logger.warning('Executing %s took %.3f seconds',
                               _format_handle(handle), dt)
        finally:
            self._current_handle = None
    else:
        handle._run()

可以看到"唯一真正执行回调的地方"就是 _run_once()，回调的调用点被刻意集中在此处以便统一做耗时统计。当你看到 Executing <Handle ...> took X.XXX seconds 这类告警时，说明某个回调/任务步长阻塞了事件循环超过 100ms，应优先排查其中的同步阻塞代码。

协程创建点（origin）追踪：debug 模式下事件循环会调用 sys.set_coroutine_origin_tracking_depth() 记录协程对象是在哪个位置创建的（见 base_events.py），这正是后面"从未被 await 的协程""Task 异常从未被取回"两类问题能在 debug 输出中附带创建栈回溯的原因。

并发与多线程：认清事件循环的线程边界

单线程协作式调度模型

事件循环运行在一个线程（通常是主线程）中，并在该线程内执行所有回调和 Task。如前所述，Task 在事件循环内运行期间，同线程的其他 Task 无法并行执行；只有当某个 Task 执行 await 时它才会挂起，事件循环转而去执行下一个 Task。

因此，千万不要把"协程并发"理解成"多线程并行"。它是单线程内的交错执行，这决定了与 asyncio 对象交互时存在严格的线程边界。

从其他 OS 线程安全调度回调：call_soon_threadsafe

要从另一个操作系统线程向事件循环调度一个回调，必须使用 loop.call_soon_threadsafe() 方法，例如：

loop.call_soon_threadsafe(callback, *args)

几乎所有 asyncio 对象都不是线程安全的。通常这不成问题——前提是你的代码只在 Task 或回调内部操作它们。如果确实需要在 Task/回调之外（例如在工作线程中）调用某个底层 asyncio API，就必须借助 call_soon_threadsafe，例如：

loop.call_soon_threadsafe(fut.cancel)

从源码看，call_soon_threadsafe()（base_events.py）除了与 call_soon() 一样把 handle 追加进 _ready 双端队列外，还会调用 self._write_to_self() 向事件循环的"自我唤醒管道"写入数据，从而确保即使事件循环正阻塞在 selector 轮询上也能被立刻唤醒，这是它能跨线程工作的关键机制。

从其他 OS 线程调度协程：run_coroutine_threadsafe

若要从另一个 OS 线程调度一个协程对象，则应使用 asyncio.run_coroutine_threadsafe() 函数。它返回一个 concurrent.futures.Future，便于等待并取回结果：

async def coro_func():
     return await asyncio.sleep(1, 42)

# Later in another OS thread:

future = asyncio.run_coroutine_threadsafe(coro_func(), loop)
# Wait for the result:
result = future.result()

注意：future.result() 会阻塞当前（工作）线程直到协程完成，适合在工作线程中等待；run_coroutine_threadsafe 本质上就是"提交协程对象 + call_soon_threadsafe 调度 + 跨线程 Future 桥接"的组合。

信号处理必须在主线程

事件循环若要处理信号（signal），必须运行在主线程中——POSIX 信号只投递给进程的主线程，这是平台层面的硬约束，代码层面无法绕过。

用 run_in_executor 执行阻塞代码

loop.run_in_executor() 方法可搭配 concurrent.futures.ThreadPoolExecutor（线程池）或 InterpreterPoolExecutor（解释器池）来执行阻塞代码，使其运行在另一个 OS 线程中，从而不阻塞事件循环所在的 OS 线程。若省略 executor 参数，事件循环会按需创建一个 thread_name_prefix='asyncio' 的默认线程池（见 base_events.py）：

def run_in_executor(self, executor, func, *args):
    self._check_closed()
    if self._debug:
        self._check_callback(func, 'run_in_executor')
    if executor is None:
        executor = self._default_executor
        # Only check when the default executor is being used
        self._check_default_executor()
        if executor is None:
            executor = concurrent.futures.ThreadPoolExecutor(
                thread_name_prefix='asyncio'
            )
            self._default_executor = executor
    ...

跨进程：目前没有直接的调度通道

目前没有任何方式直接从另一个进程（例如由 multiprocessing 启动的进程）调度协程或回调到事件循环。官方给出的可行替代方案是：

事件循环方法章节（见 Doc/library/asyncio-eventloop.rst 中的 asyncio-event-loop-methods）列出了可以从管道读取数据、监视文件描述符而不阻塞事件循环的 API；
asyncio 的子进程（Subprocess） API 可以从事件循环中启动一个进程并与之通信；
前面提到的 loop.run_in_executor() 同样可以搭配 concurrent.futures.ProcessPoolExecutor，把代码放到另一个进程里执行。

运行阻塞代码：不要让 CPU 密集逻辑饿死事件循环

阻塞（CPU 密集）代码不应该被直接调用。 举例来说，如果某个函数执行了 1 秒的 CPU 密集计算，那么事件循环中所有并发的 Task 和 I/O 操作都会被推迟整整 1 秒——因为在单线程协作模型下，事件循环无法在该函数返回前调度任何其他任务。

规避手段就是使用执行器（executor）把任务放到别的线程去跑，必要时甚至可以放到另一个解释器、另一个进程中，从而避免阻塞承载事件循环的 OS 线程。详细用法见 loop.run_in_executor() 方法说明（Doc/library/asyncio-eventloop.rst），典型写法为：

import asyncio
import concurrent.futures

def blocking_cpu_work(n):
    # CPU-intensive calculation
    return sum(i * i for i in range(n))

async def main():
    loop = asyncio.get_running_loop()
    # 放入线程池执行，事件循环不被阻塞
    result = await loop.run_in_executor(
        None, blocking_cpu_work, 10_000_000)
    print(result)

asyncio.run(main())

选择哪类执行器取决于你隔离的诉求：ThreadPoolExecutor 适合会释放 GIL 的 C 扩展阻塞调用或一般耗时操作；InterpreterPoolExecutor 能利用子解释器（sub-interpreter）实现并行并隔离状态；ProcessPoolExecutor 则适合重 CPU 计算或需要内存隔离的场景。判断依据永远是：这段代码是否会长时间占用 GIL 或阻塞 OS 线程，进而延迟同循环内的其他协程。

日志：asyncio 使用独立的 "asyncio" 日志器

asyncio 基于标准库 logging 模块做日志，所有日志都经由名为 "asyncio" 的 logger 输出。从 Lib/asyncio/log.py 可以看到其定义：

"""Logging configuration."""

import logging

# Name the logger after the package.
logger = logging.getLogger(__package__)

由于 __package__ 即 asyncio，该 logger 的名字正是 "asyncio"。默认日志级别是 logging.INFO，很容易调整，例如压掉 INFO 级噪音：

logging.getLogger("asyncio").setLevel(logging.WARNING)

一个必须警惕的实践问题是：网络日志（network logging）可能阻塞事件循环。如果日志处理器（handler）自身要写网络、写数据库或执行其他可能阻塞的 I/O，就会拖住事件循环。官方建议：

使用独立线程处理日志（如 QueueHandler + QueueListener 的组合，把日志投递到队列、由专门线程异步落盘/发送），或
使用非阻塞 I/O 的日志后端。

关于"处理会阻塞的 handler"（blocking handlers）的具体方案，可参考 Doc/howto/logging-cookbook.rst 中专门讨论排队 handler 与独立监听线程的章节。核心思路概括为一句：日志 IO 属于阻塞 IO，不应出现在事件循环线程的同步路径上。

检测从未被 await 的协程

当协程函数被调用却没有被 await（例如写了 coro() 而不是 await coro()），或者协程没有通过 asyncio.create_task() 被调度时，asyncio 会发出一个 RuntimeWarning。这是最常见的初学者错误之一，也是"静默丢任务"的头号来源。

示例：

import asyncio

async def test():
    print("never scheduled")

async def main():
    test()

asyncio.run(main())

输出：

test.py:7: RuntimeWarning: coroutine 'test' was never awaited
  test()

如果开启了调试模式，还会额外打印该协程对象的"出生地"（创建位置回溯），排查起来一目了然：

test.py:7: RuntimeWarning: coroutine 'test' was never awaited
Coroutine created at (most recent call last)
  File "../t.py", line 9, in <module>
    asyncio.run(main(), debug=True)

  < .. >

  File "../t.py", line 7, in main
    test()
  test()

这里之所以 debug 模式下能给出创建点，正是前面提到的 _set_coroutine_origin_tracking() 对 sys.set_coroutine_origin_tracking_depth() 的设置（base_events.py）在起作用。

常规修复方式有两种——要么await 这个协程，要么用 asyncio.create_task() 把它调度为后台任务：

async def main():
    await test()

经验法则：一个协程对象被创建出来，最终命运只有三种——被 await、被 create_task() 调度、被关闭（.close()）。三种都不是，那就是 bug。

检测从未被取回的异常

如果对某个 Future 调用了 Future.set_exception()，但该 Future 从未被 await，那么这个异常就永远不会传播到用户代码——等于"异常被吞掉了"。asyncio 的做法是：当这个 Future 对象被垃圾回收时，打印一条日志提醒你。

含未处理异常的经典示例：

import asyncio

async def bug():
    raise Exception("not consumed")

async def main():
    asyncio.create_task(bug())

asyncio.run(main())

输出：

Task exception was never retrieved
future: <Task finished coro=<bug() done, defined at test.py:3>
  exception=Exception('not consumed')>

Traceback (most recent call last):
  File "test.py", line 4, in bug
    raise Exception("not consumed")
Exception: not consumed

注意：main() 里创建完 Task 立即返回，事件循环随之关闭；Task 自身抛出的异常没有被任何 await/future.result() 取回（retrieved），于是任务对象在销毁时打出上述告警。你看到的 Task 正是 Future 的子类，因此这类检查同样适用于普通 Future。

启用调试模式后，输出会额外给出这个任务是在哪里创建的（此前只给得出协程定义行）：

asyncio.run(main(), debug=True)

输出：

Task exception was never retrieved
future: <Task finished coro=<bug() done, defined at test.py:3>
    exception=Exception('not consumed') created at asyncio/tasks.py:321>

source_traceback: Object created at (most recent call last):
  File "../t.py", line 9, in <module>
    asyncio.run(main(), debug=True)

  < .. >

Traceback (most recent call last):
  File "../t.py", line 4, in bug
    raise Exception("not consumed")
Exception: not consumed

修复这类问题的方法是确保每个 Task 的异常最终都被消费：要么在协程内自行 try/except，要么保存 task 引用并 await 它，要么至少为 Task 添加 done 回调来显式调用 task.exception()。

异步生成器最佳实践

编写正确高效的 asyncio 代码还要求对若干异步生成器陷阱保持警觉。本节归纳的几条最佳实践能帮你省下大量调试时间。相关 API 的权威说明见 asyncio-task（aclose、asend、athrow、anext 等异步生成器方法）与 contextlib（aclosing）文档。

实践一：显式关闭异步生成器

建议手动关闭异步生成器（asynchronous generator iterator）。原因在于：如果生成器提前退出——例如 async for 循环体内抛出了异常——那么它的异步清理代码可能在意想不到的上下文中执行：可能发生在它依赖的 Task 已经完成之后，也可能发生在事件循环关闭、异步生成器的垃圾回收钩子被调用时。

为避免这种情况，应显式调用生成器的 aclose() 方法，或使用 contextlib.aclosing() 上下文管理器：

import asyncio
import contextlib

async def gen():
  yield 1
  yield 2

async def func():
  async with contextlib.aclosing(gen()) as g:
    async for x in g:
      break  # Don't iterate until the end

asyncio.run(func())

正如上文所述，这些异步生成器的清理代码是被"延迟"的。下面的例子演示了异步生成器的终结顺序可能完全出乎意料：

import asyncio
work_done = False

async def cursor():
    try:
        yield 1
    finally:
        assert work_done

async def rows():
    global work_done
    try:
        yield 2
    finally:
        await asyncio.sleep(0.1) # imitate some async work
        work_done = True


async def main():
    async for c in cursor():
        async for r in rows():
            break
        break

asyncio.run(main())

该例子的输出为：

unhandled exception during asyncio.run() shutdown
task: <Task finished name='Task-3' coro=<<async_generator_athrow without __name__>()> exception=AssertionError()>
Traceback (most recent call last):
  File "example.py", line 6, in cursor
    yield 1
asyncio.exceptions.CancelledError

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "example.py", line 8, in cursor
    assert work_done
           ^^^^^^^^^
AssertionError

cursor() 异步生成器先于 rows 生成器被终结——这是非预期行为。直觉上 cursor 在外层、rows 在内层，应该是 rows 先清理、把 work_done 置为 True 后 cursor 的 finally 断言才能通过；但因为两者都嵌套着提前 break，其异步清理被推迟到 asyncio.run() 关闭阶段，由各自独立的收尾 Task 驱动，顺序便不可控了。

修复方式是显式关闭 cursor 和 rows 这两个异步生成器，把清理时机交还给确定性的控制流：

async def main():
    async with contextlib.aclosing(cursor()) as cursor_gen:
        async for c in cursor_gen:
            async with contextlib.aclosing(rows()) as rows_gen:
                async for r in rows_gen:
                    break
            break

用 aclosing 包住每个生成器后，rows 的 finally 会在退出内层 async with 时立即执行，cursor 的 finally 则在退出外层 async with 时执行，顺序完全可控。

实践二：仅在事件循环运行时创建异步生成器

建议只在事件循环被创建之后才创建异步生成器。

为了保证异步生成器能够可靠关闭，事件循环会借助 sys.set_asyncgen_hooks() 注册回调函数。这些回调负责把正在运行的异步生成器登记进列表，以保持状态一致。相关实现就在 Lib/asyncio/base_events.py 的 BaseEventLoop.run_forever 启动路径中：

self._set_coroutine_origin_tracking(self._debug)
self._old_agen_hooks = sys.get_asyncgen_hooks()
sys.set_asyncgen_hooks(
    firstiter=self._asyncgen_firstiter_hook,
    finalizer=self._asyncgen_finalizer_hook
)

两个钩子分别负责把生成器加入、移出事件循环的 _asyncgens 弱引用集合（base_events.py）：

def _asyncgen_finalizer_hook(self, agen):
    self._asyncgens.discard(agen)

def _asyncgen_firstiter_hook(self, agen):
    if self._asyncgens_shutdown_called:
        ...
    self._asyncgens.add(agen)

当调用 loop.shutdown_asyncgens() 时（asyncio.run() 关闭阶段会自动调用），正在运行的生成器会被优雅停止、列表被清空。其实现（base_events.py）是对所有存活生成器并发执行 aclose()：

async def shutdown_asyncgens(self):
    """Shutdown all active asynchronous generators."""
    self._asyncgens_shutdown_called = True

    if not len(self._asyncgens):
        # If Python version is <3.6 or we don't have any asynchronous
        # generators alive.
        return

    closing_agens = list(self._asyncgens)
    self._asyncgens.clear()

    results = await tasks.gather(
        *[ag.aclose() for ag in closing_agens],
        return_exceptions=True)

    for result, agen in zip(results, closing_agens):
        if isinstance(result, BaseException):
            self.call_exception_handler({
                'message': f'an error occurred during closing of '
                           f'asynchronous generator {agen!r}',
                'exception': result,
                'asyncgen': agen
            })

关键时序在于：异步生成器在第一次迭代时才调用对应的系统钩子，同时生成器会记录"钩子已调用"并保证不再重复调用。因此，如果迭代发生在事件循环创建之前，由于此时钩子尚未被事件循环安装，生成器调用钩子会落空，事件循环无法把它加入活跃生成器列表，自然也就无法在必要时将其终止。

下面的失败示例展示了一个常见误区——用 asyncio.Runner 时在 runner 运行前就先对生成器做首次 anext：

import asyncio

async def agenfn():
    try:
        yield 10
    finally:
        await asyncio.sleep(0)


with asyncio.Runner() as runner:
    agen = agenfn()
    print(runner.run(anext(agen)))
    del agen

输出：

10
Exception ignored while closing generator <async_generator object agenfn at 0x000002F71CD10D70>:
Traceback (most recent call last):
  File "example.py", line 13, in <module>
    del agen
        ^^^^
  RuntimeError: async generator ignored GeneratorExit

这个例子可以这样修复——把首次迭代挪进 main() 协程，保证事件循环已经运行：

import asyncio

async def agenfn():
    try:
        yield 10
    finally:
        await asyncio.sleep(0)

async def main():
    agen = agenfn()
    print(await anext(agen))
    del agen

asyncio.run(main())

实践三：避免对同一个生成器并发迭代与关闭

异步生成器可能在一个 __anext__() / athrow() / aclose() 调用尚未结束时就再次被进入（reentered），这会把生成器带入不一致状态并引发错误。

看下面的例子——两个 Task 同时对同一个生成器调用 asend()：

import asyncio

async def consumer():
    for idx in range(100):
        await asyncio.sleep(0)
        message = yield idx
        print('received', message)

async def amain():
    agenerator = consumer()
    await agenerator.asend(None)

    fa = asyncio.create_task(agenerator.asend('A'))
    fb = asyncio.create_task(agenerator.asend('B'))
    await fa
    await fb

asyncio.run(amain())

输出：

received A
Traceback (most recent call last):
  File "test.py", line 38, in <module>
    asyncio.run(amain())
    ~~~~~~~~~~~^^^^^^^^^
  File "Lib/asyncio/runners.py", line 204, in run
    return runner.run(main)
           ~~~~~~~~~~^^^^^^
  File "Lib/asyncio/runners.py", line 127, in run
    return self._loop.run_until_complete(task)
           ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~^^^^^^
  File "Lib/asyncio/base_events.py", line 719, in run_until_complete
    return future.result()
           ~~~~~~~~~~~~~^^
  File "test.py", line 36, in amain
    await fb
RuntimeError: anext(): asynchronous generator is already running

栈回溯中 Lib/asyncio/runners.py 与 Lib/asyncio/base_events.py 的行号与本仓库 Lib/asyncio/runners.py 的 run()/run_until_complete() 调用链吻合——这正是 asyncio.run() 经 Runner.run 再到 loop.run_until_complete 的完整执行路径。

因此官方建议很明确：避免在并行 Task 中、或在多个事件循环之间并发使用同一个异步生成器。如果确实需要"并发生成数据"，请为每个消费者创建独立的生成器实例，而不是共享一个。

总结：asyncio 开发的黄金检查清单

关注点	检查与行动
调试	开发期用 `asyncio.run(main(), debug=True)` 或 `PYTHONASYNCIODEBUG=1`；必要时 `python -W default -X dev` 叠加资源告警
慢回调	关注 `Executing <Handle> took X.XXX seconds` 告警；按需调整 `loop.slow_callback_duration`
跨线程	回调用 `loop.call_soon_threadsafe()`，协程用 `asyncio.run_coroutine_threadsafe()`，绝不直接操作非线程安全的 asyncio 对象
阻塞代码	CPU 密集/阻塞调用一律交 `loop.run_in_executor()`（线程池/解释器池/进程池），不要裸调
日志	只调 `"asyncio"` logger；避免在事件循环线程做阻塞式网络日志，采用队列 + 独立线程
未等待协程	若见 `RuntimeWarning: coroutine ... was never awaited`，补 `await` 或 `create_task()`
未取回异常	若见 `Task exception was never retrieved`，确保每个 Task 的异常都被消费
异步生成器	用 `contextlib.aclosing()` 显式关闭；仅在事件循环运行时开始迭代；绝不并发迭代/关闭同一生成器

所有示例与告警机制均可对照 CPython 仓库源码进一步验证：调试模式的实现集中在 Lib/asyncio/base_events.py（set_debug、slow_callback_duration、_run_once、_check_thread、asyncgen 钩子与 shutdown_asyncgens）；PYTHONASYNCIODEBUG 与开发模式读取在 Lib/asyncio/coroutines.py；"asyncio" logger 在 Lib/asyncio/log.py。若需回归验证官方行为，可在 Lib/test 中检索 test_asyncio 相关测试用例（如 test_tasks.py、test_base_events.py 对 slow callback、never-awaited coroutine、asyncgen shutdown 的断言）。记住一句话：调试模式与显式资源管理，是让 asyncio 从"能跑"走向"可靠"的两大杠杆。