如何利用Generator实现无asyncio的异步图片下载器并修复同步问题

你的代码目前是同步执行的核心原因：download_file是普通函数，调用时会完整执行完所有逻辑才会返回，生成器async_downloader每次yield的是函数执行结果而非暂停点，导致任务逐个串行执行。要实现基于生成器的异步，必须将阻塞操作拆分为可暂停的生成器步骤，让调度器能在多个任务间切换控制权。

修复后的代码 #

import time
import urllib.request

delays = iter((4, 1, 2, 3, 4, 5))

def download_file(url, filename):
    delay = next(delays)
    # 模拟阻塞等待，通过yield让出控制权给调度器
    for _ in range(delay):
        yield
        time.sleep(1)
    # 下载前再次让出控制权
    yield
    with urllib.request.urlopen(url) as response, open(filename, 'wb') as out_file:
        data = response.read()
        out_file.write(data)
    print(f'{filename} downloaded, delay {delay}')

def run_async(tasks):
    # 协作式调度器：轮询推进所有活跃任务
    while tasks:
        remaining_tasks = []
        for task in tasks:
            try:
                # 推进生成器到下一个yield点
                next(task)
                remaining_tasks.append(task)
            except StopIteration:
                # 任务完成，不再加入下一轮
                pass
        tasks = remaining_tasks

FILES = [
    ['http://farm4.static.flickr.com/3488/4051378654_238ca94313.jpg', 'n00004475_42770.jpg'],
    ['http://farm4.static.flickr.com/3368/3198142470_6eb0be5f32.jpg', 'n00004475_54295.jpg'],
    ['http://farm4.static.flickr.com/3276/2875184020_9944005d0d.jpg', 'n00004475_15899.jpg'],
    ['http://farm3.static.flickr.com/2531/4094333885_e8462a8338.jpg', 'n00004475_32312.jpg'],
    ['http://farm4.static.flickr.com/3289/2809605169_8efe2b8f27.jpg', 'n00004475_35466.jpg']
]

# 创建所有下载任务的生成器实例
download_tasks = [download_file(url, filename) for url, filename in FILES]
# 启动调度器
run_async(download_tasks)

核心修改说明 #

1. 将download_file改为生成器：

   • 把原本一次性的time.sleep(delay)拆分为多次yield+time.sleep(1)，每次yield都会暂停当前任务，把控制权交还给调度器，让其他任务有机会执行。
   • 在实际下载操作前也加入yield，进一步拆分阻塞点。

2. 实现协作式调度器run_async：

   • 维护一个活跃任务列表，每次循环遍历所有任务，调用next()推进生成器到下一个暂停点。
   • 当任务抛出StopIteration时，说明任务已完成，将其从活跃列表中移除。

3. 利用yield实现任务切换：
   生成器的yield是协作式多任务的核心——任务主动让出控制权，调度器负责轮询所有任务，实现类似异步的并行执行效果。

进阶：结合yield from优化调度 #

如果想使用yield from简化子生成器的调用，可以调整调度逻辑：

def async_downloader(files):
    tasks = [download_file(url, filename) for url, filename in files]
    while tasks:
        for task in list(tasks):
            try:
                # yield from会自动推进子生成器直到下一个yield点
                yield from task
            except StopIteration:
                tasks.remove(task)

def run_scheduler(files):
    gen = async_downloader(files)
    while True:
        try:
            next(gen)
        except StopIteration:
            break

# 调用方式
run_scheduler(FILES)

yield from可以自动处理子生成器的yield和StopIteration，让主生成器的逻辑更简洁，但核心的协作式切换逻辑和之前一致。

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