优化asyncio中嵌套异步任务的并发调度

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():
    i=0
    async for sentence in sentences_generator():
        print("processing sentence: ", i)
        await process_sentence(sentence) # 这里的await导致阻塞
        i += 1

# asyncio.run(main())

got char: H
got char: i
got char: .
got sentence:  Hi.
processing sentence:  0
waiting for processing sentence:  Hi.
sentence processed!
got char:  
got char: H
got char: e
got char: y
got char: .
got sentence:   Hey.
processing sentence:  1
waiting for processing sentence:   Hey.
sentence processed!
...

import asyncio

# 定义全局变量用于计数，方便观察
i = 1

async def stream():
    char_string = "Hi. Hello. Thank you." # 增加一些内容以更好地展示并发
    for char in char_string:
        await asyncio.sleep(0.1) # 模拟耗时操作
        print("got char:", char)
        yield char

async def sentences_generator(q: asyncio.Queue[str], flag: asyncio.Event):
    """
    生产者协程：从字符流生成句子，并放入队列。
    当所有句子生成完毕后，设置flag通知消费者。
    """
    sentence = ""
    async for char in stream():
        sentence += char
        if char in [".", "!", "?"]:
            print("got sentence: ", sentence)
            await q.put(sentence) # 将生成的句子放入队列
            sentence = ""
    # 确保最后一个不以标点符号结尾的句子也被处理（如果需要）
    if sentence:
        print("got sentence: ", sentence)
        await q.put(sentence)
    flag.set() # 生产完毕，设置事件标志

async def process_sentence(q: asyncio.Queue[str], flag: asyncio.Event):
    """
    消费者协程：从队列中获取句子并进行处理。
    当队列为空且生产者已设置flag时，停止消费。
    """
    global i
    while True:
        # 检查是否应该停止：队列为空且生产者已完成
        if q.empty() and flag.is_set():
            break

        # 尝试从队列获取项目，如果队列为空则等待
        item = await q.get()

        print("processing sentence: ", i)
        print("waiting for processing sentence: ", item)
        await asyncio.sleep(len(item) * 0.1) # 模拟耗时处理
        print("sentence processed!")

        q.task_done() # 通知队列此任务已完成
        i += 1

async def main():
    global i
    i = 1 # 重置计数器
    event = asyncio.Event() # 用于生产者通知消费者结束
    queue = asyncio.Queue[str]() # 生产者和消费者之间的通信队列

    # 启动生产者和消费者作为独立的协程任务
    producer_task = asyncio.create_task(sentences_generator(queue, event))
    consumer_task = asyncio.create_task(process_sentence(queue, event))

    # 等待所有任务完成
    await asyncio.gather(producer_task, consumer_task)

    # 可选：等待队列中所有任务被标记为完成，确保所有数据都被处理
    await queue.join()

asyncio.run(main())

got char: H
got char: i
got char: .
got sentence:  Hi.
got char:  
got char: H
processing sentence:  1
waiting for processing sentence:  Hi.
got char: e
got char: l
got char: l
got char: o
got char: .
got sentence:   Hello.
sentence processed!
got char:  
got char: T
processing sentence:  2
waiting for processing sentence:   Hello.
got char: h
got char: a
got char: n
got char: k
got char:  
got char: y
got char: o
got char: u
got char: .
got sentence:  Thank you.
sentence processed!
processing sentence:  3
waiting for processing sentence:  Thank you.
sentence processed!