Filas & Produtor/Consumidor
Pipelines de produtor/consumidor desacoplam estágios com queue.Queue (threads) ou multiprocessing.Queue (processos) de tamanho limitado. Produtores enfileiram trabalho; consumidores processam em seu próprio ritmo - suavizando picos e simplificando o backpressure.
Receita
from queue import Queue
from threading import Thread
q: Queue[str] = Queue(maxsize=10)
def producer() -> None:
for i in range(5):
q.put(f"job-{i}")
def consumer() -> None:
while True:
item = q.get()
try:
print("process", item)
finally:
q.task_done()
Thread(target=consumer, daemon=True).start()
producer()
q.join()Quando usar isso:
- Pipelines de download -> análise -> armazenamento
- Pools de threads alimentados por um único produtor
- Backpressure via bloqueio
putcommaxsize - Desligamento gracioso com sentinelas ou
join - Workers de log/métricas fora do caminho crítico
Exemplo de Trabalho
from queue import Queue
from threading import Thread, Event
STOP = object()
def worker(q: Queue, stop_event: Event) -> None:
while not stop_event.is_set():
try:
item = q.get(timeout=0.1)
except Exception:
continue
if item is STOP:
q.task_done()
break
print("handled", item)
q.task_done()
q: Queue = Queue(maxsize=3)
stop_event = Event()
threads = [Thread(target=worker, args=(q, stop_event)) for _ in range(2)]
for t in threads:
t.start()
for i in range(6):
q.put(i)
for _ in threads:
q.put(STOP)
stop_event.set()
q.join()
for t in threads:
t.join()O que isso demonstra:
- Fila limitada bloqueia o produtor quando cheia (
maxsize=3) task_done/joinrastreiam o trabalho pendente- Sentinela STOP por consumidor para saída limpa
- Múltiplos consumidores competem de forma justa no
get
Mergulho Profundo
Tipos de Fila
| Fila | Escopo |
|---|---|
queue.Queue | Threads no mesmo processo |
multiprocessing.Queue | IPC entre processos |
asyncio.Queue | Corrotinas assíncronas |
Backpressure
put(block=True, timeout=...)quando cheia- Monitore
qsize()com cautela - aproximado em multiprocessing
Armadilhas
task_doneesquecido -jointrava para sempre. Correção:try/finally: q.task_done()por item.- Memória de fila ilimitada - produtores superam consumidores. Correção:
maxsize+ put bloqueante. - Pílula de veneno por consumidor - uma única sentinela para apenas uma thread. Correção: um STOP por worker.
- Pickling de objetos grandes em filas MP - lento. Correção: memória compartilhada ou caminhos para arquivos.
- Processamento dentro de
getsem timeout - não é possível interromper o desligamento. Correção: timeouts + evento.
Alternativas
| Alternativa | Use Quando | Não Use Quando |
|---|---|---|
| asyncio.Queue | Pipeline assíncrono | Consumidores bloqueantes |
| Broker externo (Redis) | Entre serviços | Apenas dentro do processo |
| Chunks de iterador | Scripts de lote simples | Muitos produtores |
FAQs
Significado de maxsize=0?
Fila ilimitada - put nunca bloqueia por capacidade.
PriorityQueue?
queue.PriorityQueue para trabalhos ordenados - tuplas (prioridade, item).
Quantos consumidores?
Corresponde à taxa de transferência sustentada - frequentemente semelhante ao limite de paralelismo de I/O.
Diferenças do asyncio.Queue?
Await put/get; não é thread-safe com threading sem uma ponte.
Como testar pipelines?
Use maxsize pequeno, itens determinísticos, join com timeout nos testes.
Fila de mensagens mortas (dead letter queue)?
Fila secundária para itens falhados após N tentativas - padrão sobre a Fila base.
Produtor de logging?
QueueHandler em logging roteia registros para uma thread de listener - padrão da biblioteca padrão.
Justiça (Fairness)?
Fila padrão é FIFO; múltiplos consumidores obtêm distribuição aproximadamente justa.
Fila de processo vs thread?
Nunca misture - escolha a fila de threading ou multiprocessing que corresponda aos workers.
Ordem de desligamento?
Pare os produtores primeiro, drene a fila, envie sentinelas, junte os consumidores.
Relacionado
- threading - threads de worker
- multiprocessing - filas de processo
- concurrent.futures - alternativa de pool
- Asyncio Basics - asyncio.Queue
Versões do Stack: Esta página foi escrita para Python 3.14.0 (estável 3.14, manutenção 3.13), FastAPI 0.115+, Django 5.2, Flask 3.1, Pydantic 2, PyTorch 2.6+, pandas 2.2+, Polars 1.x, ruff 0.9+, e uv 0.6+.