优化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: l
got char: l
got char: o
got char: .
got sentence:   Hello.
processing sentence:  1
waiting for processing sentence:   Hello.
sentence processed!

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(q: asyncio.Queue[str], flag: asyncio.Event):
    """
    生产者：从字符流生成句子，并放入队列。
    当字符流结束时，设置Event标志通知消费者。
    """
    sentence = ""
    async for char in stream():
        sentence += char
        if char in [".", "!", "?"]:
            print("got sentence: ", sentence)
            await q.put(sentence) # 将生成的句子放入队列
            sentence = ""
    flag.set() # 生产者完成所有句子的生成，设置Event

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 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)

asyncio.run(main())

got char: H
got char: i
got char: .
got sentence:  Hi.
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:  Hi.
got char:  
got char: T
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:  2
waiting for processing sentence:   Hello.
sentence processed!
processing sentence:  3
waiting for processing sentence:   Thank you.
sentence processed!