如何优化asyncio嵌套函数调度，实现句子处理与字符流读取并发？

原代码的核心问题是：在main函数中，每次调用await process_sentence(sentence)会阻塞当前协程，直到句子处理完成才会继续从sentences_generator获取下一个句子，导致字符流的生成被暂停。要实现"处理句子的同时继续获取后续字符"的效果，需要把句子处理操作变成后台任务，让主协程不用等待它完成就能继续执行。

修改后的代码 #

import asyncio


async def stream():
    char_string = "Hi. Hello. Hello."

    for char in char_string:
        await asyncio.sleep(0.1)  # 模拟耗时字符获取
        print("got char:", char)
        yield char


async def sentences_generator():
    sentence = ""
    async for char in stream():
        sentence += char
        if char in [".", "!", "?"]:
            print("got sentence: ", sentence)
            yield sentence
            sentence = ""


async def process_sentence(sentence: str):
    print("waiting for processing sentence: ", sentence)
    await asyncio.sleep(len(sentence)*0.1)
    print("sentence processed!")


async def main():
    tasks = []
    i = 0
    async for sentence in sentences_generator():
        print("processing sentence: ", i)
        # 创建后台任务，不阻塞主协程
        task = asyncio.create_task(process_sentence(sentence))
        tasks.append(task)
        i += 1
    # 等待所有后台任务处理完成
    await asyncio.gather(*tasks)

asyncio.run(main())

关键修改点 #

1. 用asyncio.create_task创建后台任务：
   原代码中await process_sentence(sentence)会让主协程暂停，直到该函数执行完毕。改用create_task后，句子处理会在后台独立运行，主协程可以立刻继续执行async for循环，继续从字符流中获取后续字符。
2. 收集所有任务并最终等待完成：
   如果不等待后台任务，main函数可能会在所有句子处理完成前就结束，导致程序退出。因此需要把所有创建的任务存入列表，最后用asyncio.gather(*tasks)等待全部任务完成。

输出效果 #

修改后的代码会输出类似你期望的结果：字符流的获取和句子处理会并发进行，比如在处理第一个句子Hi.的同时，程序会继续获取后续的空格、H等字符，实现并行执行的效果。

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