Использование asyncio.Queue для потока производитель-потребитель ⇐ Python

[Anonymous]

Я не понимаю, как использовать asyncio.Queue для конкретного шаблона производитель-потребитель, в котором производитель и потребитель работают одновременно и независимо.

Во-первых, рассмотрим этот пример, который точно повторяет пример из документации для 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 в качестве сигнала потребителю о том, что производство завершено.

Подробнее здесь: https://stackoverflow.com/questions/525 ... sumer-flow