Asyncio.Queue – есть ли преимущества в асинхронном использовании с maxsize? ⇐ Python

[Anonymous]

У меня есть функция вызова синхронизации, которая помещает задачу в asyncio.Queue, чтобы ее мог обработать асинхронный исполнитель. Сейчас у меня есть что-то вроде этого:

Код: Выделить всё

import asyncio
import os
import sys

queue = asyncio.Queue(maxsize=0)

async def worker():
try:
while True:
task = await queue.get()
print(f'processing task: {task}')
queue.task_done()
except asyncio.QueueShutDown:
return

def on_read():
data = os.read(0, 10)
if data:
queue.put_nowait(data)
else:
queue.shutdown()

loop = asyncio.new_event_loop()
loop.add_reader(0, on_read)
loop.run_until_complete(worker())

Код можно использовать следующим образом:

Код: Выделить всё

for i in $(seq 3); do echo $i; sleep 1; done | python3 minimal_queue.py

Обратите внимание, что я использую put_nowait для помещения задачи в очередь. Теоретически это может привести к увеличению QueueFull, но поскольку очередь неограничена (

Код: Выделить всё

maxsize=0

) этого никогда не должно произойти. Однако это может означать, что очередь растет до тех пор, пока в системе не закончится память.
Я думал об изменении этого параметра, чтобы установить максимальный размер очереди, а затем использовать цикл.create_task(queue.put(data)). Но я не уверен в этом. Я предполагаю, что в условиях стресса это просто поставит в очередь массу заблокированных задач, которые могут заполнить память еще быстрее.
Третий вариант — использовать put_nowait() и по-прежнему устанавливать максимальный размер. Это приведет к сбою гораздо раньше, но не съест всю память в системе и, следовательно, окажет меньшее влияние на другие процессы в системе.
Сейчас я склоняюсь к третьему варианту. Но я не уверен, что до конца понимаю, как работают эти очереди. Верны ли мои рассуждения?

Подробнее здесь: https://stackoverflow.com/questions/798 ... th-maxsize