自定义Python协程实现（无asyncio）：异步机制底层理解与问题排查

嘿，我完全懂你想手动实现Python协程、不靠asyncio来吃透异步底层的想法——这绝对是搞懂并发机制最扎实的路子！先从你提到的同步stacker示例说起，咱们一步步拆解问题，再把你异步实现里的bug揪出来。

同步版Stacker实现与预期输出 #

先明确同步场景下的行为，比如这个典型的stacker例子：

import time

def stacker(n):
    for i in range(n):
        print(f"Stacker {i}")
        # 模拟等待/无操作状态
        time.sleep(0.5)

# 同步执行两个任务
stacker(3)
stacker(2)

预期输出（严格按顺序执行，先跑完第一个任务再跑第二个）：

Stacker 0
Stacker 1
Stacker 2
Stacker 0
Stacker 1

你的异步实现可能踩的坑 #

手动实现协程核心是用**生成器（yield）**做上下文切换，但很多人会犯这几个错误：

• 用了time.sleep()这种阻塞调用：它会卡住整个进程，而不是只让当前协程暂停，根本没给其他任务切换的机会
• 没有调度器管理任务：单独遍历生成器还是会按顺序执行，必须有一个“调度器”来轮流恢复协程
• 协程没主动让出控制权：在等待点没写yield，协程会一直跑到结束，不会给其他任务留时间

比如你可能写了类似这样的错误代码：

# 错误的异步实现示例
import time

def async_stacker(n):
    for i in range(n):
        print(f"Async Stacker {i}")
        time.sleep(0.5)  # 阻塞整个进程，无法切换
        yield  # 但yield没被调度器处理

# 错误的执行方式
gen1 = async_stacker(3)
gen2 = async_stacker(2)
for _ in gen1:
    pass
for _ in gen2:
    pass

这个代码的输出还是同步的，完全达不到你要的并发效果。

正确的手动协程实现（无asyncio） #

我们用生成器做协程载体，在等待点用yield让出控制权，再写一个极简调度器来管理任务队列，实现切换：

def async_stacker(name, n):
    for i in range(n):
        print(f"{name} {i}")
        # 模拟等待操作：主动让出控制权，告诉调度器可以切换到其他任务
        yield

# 极简调度器（手动实现的迷你事件循环）
def scheduler(tasks):
    while tasks:
        # 取出队列第一个任务
        task = tasks.pop(0)
        try:
            # 恢复协程执行，直到下一个yield
            next(task)
            # 协程还没结束，放回队列末尾等待下一次调度
            tasks.append(task)
        except StopIteration:
            # 协程执行完毕，跳过
            pass

# 创建两个协程任务
task1 = async_stacker("Stacker A", 3)
task2 = async_stacker("Stacker B", 2)

# 启动调度器
scheduler([task1, task2])

预期输出（并发效果）：
两个任务交替执行，完美实现“第一个任务等待时启动第二个任务”的需求：

Stacker A 0
Stacker B 0
Stacker A 1
Stacker B 1
Stacker A 2

进阶：模拟真实IO等待 #

如果要更贴近真实场景（比如等待网络请求），可以给协程加等待时间，让调度器处理延迟任务：

import time

def async_stacker(name, n):
    for i in range(n):
        print(f"{name} {i}")
        # 模拟IO等待：记录等待结束时间
        wait_until = time.time() + 0.5
        yield wait_until  # 把等待时间传给调度器

# 支持延迟任务的调度器
def scheduler(tasks):
    while tasks:
        # 按等待时间排序，优先处理最早到期的任务
        tasks.sort(key=lambda x: x[1])
        task, wait_until = tasks.pop(0)
        
        # 等待到任务到期（模拟IO等待）
        current_time = time.time()
        if current_time < wait_until:
            time.sleep(wait_until - current_time)
        
        try:
            # 恢复协程，获取下一个等待时间
            next_wait = next(task)
            tasks.append((task, next_wait))
        except StopIteration:
            pass

# 初始化任务队列
task1 = async_stacker("Stacker A", 3)
task2 = async_stacker("Stacker B", 2)
tasks = [(task1, next(task1)), (task2, next(task2))]

scheduler(tasks)

这个版本的输出和之前一致，但更贴近真实异步IO的工作逻辑——调度器会在任务等待时，去处理其他就绪的任务。

内容的提问来源于stack exchange，提问作者srjjio