运行 Mini Asyncio

这一章运行 mini_asyncio 的测试和 Demo，再对照标准库 asyncio 看它少做了哪些事。

运行测试

python3 -m unittest discover -s examples/mini_asyncio_runtime/tests

测试覆盖：

• Task、sleep 和返回值。
• gather 按输入顺序收集结果。
• AsyncQueue.put/get/task_done/join。
• socketpair 上的 sock_recv、sock_send_all。
• 取消向协程注入 CancelledError。

运行 Demo

python3 examples/mini_asyncio_runtime/demos/01_tasks.py
python3 examples/mini_asyncio_runtime/demos/02_gather.py
python3 examples/mini_asyncio_runtime/demos/03_queue_workers.py
python3 examples/mini_asyncio_runtime/demos/04_socket_echo.py

从一个例子开始读

从这个例子开始：

async def child(name, delay):
    await aio.sleep(delay)
    return name

async def main():
    task_a = aio.create_task(child("A", 0.2))
    task_b = aio.create_task(child("B", 0.1))
    print(await task_a)
    print(await task_b)

这段代码会这样走：

它和标准库差在哪里

功能	mini_asyncio	标准库 asyncio
事件循环	单一教学实现	多平台实现、Runner、policy 兼容
Task	让 coroutine 继续执行	取消计数、contextvars、调试信息
Future	基础状态和 callback	C 加速、更多边角检查
I/O	socket read/write	streams、subprocess、DNS、SSL、pipes
一组任务的收尾	未实现 TaskGroup	TaskGroup、ExceptionGroup
Queue	unbounded queue	maxsize、PriorityQueue、LifoQueue、shutdown
调试	无	debug mode、慢 callback、资源警告

差距很大，但最小骨架一致：某个等待源完成 Future，Future 叫醒 Task，Task 让 coroutine 继续执行。

结课检查

如果你能回答下面问题，asyncio 的运行方式基本就通了：

1. 为什么 async def 调用后不会立刻执行？
2. Task._step 为什么要调用 coro.send()？
3. await sleep() 为什么不会堵住线程？
4. Future 的 done callback 为什么要进入 ready 队列？
5. Queue 的 get() 在没有 item 时为什么能暂停当前 Task？
6. selector I/O 如何把 socket 可读事件变成 Future 结果？
7. 标准库 TaskGroup 比教学版多处理了哪些问题？