优化 asyncio 任务调度：使用队列实现生产者-消费者模式

import asyncio

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():
    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) # 这里会阻塞
        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: l
got char: l
got char: o
got char: .
got sentence:   Hello.
processing sentence:  1
waiting for processing sentence:   Hello.
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):
    """
    生产者：从字符流生成句子，并放入队列。
    当所有句子生成完毕后，设置事件标志。
    """
    sentence = ""
    async for char in stream():
        sentence += char
        if char in [".", "!", "?"]:
            print("got sentence: ", sentence)
            await q.put(sentence) # 将句子放入队列
            sentence = ""
    flag.set() # 生产者完成所有工作，设置事件标志

async def process_sentence(q: asyncio.Queue[str], flag: asyncio.Event):
    """
    消费者：从队列中取出句子并处理。
    当队列为空且生产者已完成时，退出。
    """
    global i # 使用全局计数器
    while True:
        # 检查退出条件：队列为空且生产者已设置完成标志
        if q.empty() and flag.is_set():
            break
        try:
            item = await asyncio.wait_for(q.get(), timeout=1.0) # 尝试从队列获取，设置超时避免无限等待
        except asyncio.TimeoutError:
            # 如果超时，再次检查退出条件，防止生产者完成但队列仍空的情况
            if q.empty() and flag.is_set():
                break
            continue # 继续尝试获取

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

async def main():
    global i
    i = 1 # 重置计数器
    event = asyncio.Event() # 创建事件对象
    queue = asyncio.Queue[str]() # 创建队列对象

    # 创建生产者和消费者任务
    producer_task = sentences_generator(queue, event)
    consumer_task = process_sentence(queue, event)

    # 并发运行生产者和消费者任务
    await asyncio.gather(producer_task, consumer_task)

if __name__ == "__main__":
    asyncio.run(main())

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