代码之家 › 专栏 › 技术社区 › Brad Solomon

为生产者-消费者流使用asyncio.Queue

python-asyncio async-await python-3.x python

Brad Solomon · 技术社区 · 8 年前

我不知道怎么用 asyncio.Queue 对于生产者和消费者同时独立运作的特定生产者-消费者模式。

首先,考虑这个例子,它紧跟着 docs for asyncio.Queue :

import asyncio
import random
import time

async def worker(name, queue):
    while True:
        sleep_for = await queue.get()
        await asyncio.sleep(sleep_for)
        queue.task_done()
        print(f'{name} has slept for {sleep_for:0.2f} seconds')

async def main(n):
    queue = asyncio.Queue()
    total_sleep_time = 0
    for _ in range(20):
        sleep_for = random.uniform(0.05, 1.0)
        total_sleep_time += sleep_for
        queue.put_nowait(sleep_for)
    tasks = []
    for i in range(n):
        task = asyncio.create_task(worker(f'worker-{i}', queue))
        tasks.append(task)
    started_at = time.monotonic()
    await queue.join()
    total_slept_for = time.monotonic() - started_at
    for task in tasks:
        task.cancel()
    # Wait until all worker tasks are cancelled.
    await asyncio.gather(*tasks, return_exceptions=True)
    print('====')
    print(f'3 workers slept in parallel for {total_slept_for:.2f} seconds')
    print(f'total expected sleep time: {total_sleep_time:.2f} seconds')

if __name__ == '__main__':
    import sys
    n = 3 if len(sys.argv) == 1 else sys.argv[1]
    asyncio.run(main())

关于这个脚本有一个更详细的细节:项目被同步地放入队列中,并且 queue.put_nowait(sleep_for) 在传统的for循环上。

我的目标是创建一个使用 async def worker() consumer() )以及 async def producer() . 两者都应安排为同时运行。没有一个消费者协同程序被明确地绑定到一个生产者或与之链接。

我如何修改上面的程序,使生产者是它自己的协同程序,可以与使用者/工作者同时调度?

_{第二个例子来自

PYMOTW

. 它要求生产者提前知道消费者的数量,并使用

None

作为向消费者发出的生产已经完成的信号。}

1 回复 | 直到 8 年前

user4815162342 6 年前

我如何修改上面的程序,使生产者是它自己的协同程序,可以与使用者/工作者同时调度?

将插入循环移动到单独的生产者协程。
在后台启动消费者,让他们在产品生产时处理产品。
在消费者运行的情况下,启动生产者并等待他们完成生产项目,如 await producer() 或 await gather(*producers)
一旦所有生产商都完成了,等待消费者用 await queue.join()

下面是实现上述功能的示例:

import asyncio, random
 
async def rnd_sleep(t):
    # sleep for T seconds on average
    await asyncio.sleep(t * random.random() * 2)
 
async def producer(queue):
    while True:
        # produce a token and send it to a consumer
        token = random.random()
        print(f'produced {token}')
        if token < .05:
            break
        await queue.put(token)
        await rnd_sleep(.1)
 
async def consumer(queue):
    while True:
        token = await queue.get()
        # process the token received from a producer
        await rnd_sleep(.3)
        queue.task_done()
        print(f'consumed {token}')
 
async def main():
    queue = asyncio.Queue()
 
    # fire up the both producers and consumers
    producers = [asyncio.create_task(producer(queue))
                 for _ in range(3)]
    consumers = [asyncio.create_task(consumer(queue))
                 for _ in range(10)]
 
    # with both producers and consumers running, wait for
    # the producers to finish
    await asyncio.gather(*producers)
    print('---- done producing')
 
    # wait for the remaining tasks to be processed
    await queue.join()
 
    # cancel the consumers, which are now idle
    for c in consumers:
        c.cancel()
 
asyncio.run(main())

请注意,在实际的生产者和消费者中,尤其是那些涉及网络访问的生产者和消费者,您可能希望捕获处理过程中发生的与IO相关的异常。如果异常是可恢复的,就像大多数与网络相关的异常一样,您可以简单地捕获异常并记录错误。你还是应该调用 task_done() queue.join() 将由于未处理的项目而挂起。如果重新尝试处理该项是合理的,则可以在调用之前将其返回到队列中 任务完成() . 例如:

# like the above, but handling exceptions during processing:
async def consumer(queue):
    while True:
        token = await queue.get()
        try:
            # this uses aiohttp or whatever
            await process(token)
        except aiohttp.ClientError as e:
            print(f"Error processing token {token}: {e}")
            # If it makes sense, return the token to the queue to be
            # processed again. (You can use a counter to avoid
            # processing a faulty token infinitely.)
            #await queue.put(token)
        queue.task_done()
        print(f'consumed {token}')