queue no Python: coordene threads

O módulo queue no Python oferece filas sincronizadas para trocar tarefas e resultados com segurança entre várias threads. Ele encapsula locks e condições, permitindo que produtores aguardem espaço e consumidores aguardem trabalho sem implementar sincronização manual.

Uma fila não torna a tarefa paralela por si só. Ela organiza a comunicação, aplica ordem e pode criar backpressure. O ganho depende do tipo de carga: threads são úteis principalmente em entrada e saída, como rede, arquivos e APIs. Para entender a base, veja o guia de threading no Python.

Queue FIFO

Queue entrega primeiro o item inserido primeiro:

from queue import Queue

fila = Queue(maxsize=100)
fila.put({"id": 1, "url": "https://exemplo.com"})
tarefa = fila.get()
try:
    processar(tarefa)
finally:
    fila.task_done()

maxsize limita itens pendentes. Quando a fila está cheia, put() bloqueia até surgir espaço. Isso evita que produtores rápidos consumam memória indefinidamente.

Produtor e consumidor

import queue
import threading

fila = queue.Queue(maxsize=50)

def consumidor():
    while True:
        try:
            tarefa = fila.get(timeout=1)
        except queue.Empty:
            continue
        try:
            executar(tarefa)
        except Exception as erro:
            registrar_falha(tarefa, erro)
        finally:
            fila.task_done()

threads = [
    threading.Thread(target=consumidor, daemon=True)
    for _ in range(4)
]
for thread in threads:
    thread.start()

for tarefa in carregar_tarefas():
    fila.put(tarefa, timeout=5)

fila.join()

Cada get() precisa de exatamente um task_done(), inclusive quando o processamento falha. Coloque-o em finally. join() aguarda o contador de tarefas não finalizadas chegar a zero.

Backpressure com maxsize

Uma fila ilimitada pode crescer até esgotar memória. Escolha maxsize conforme custo médio da tarefa, quantidade de workers e tolerância de espera. Quando a fila estiver cheia, o produtor pode bloquear, desistir ou registrar a tarefa em armazenamento persistente.

import queue

try:
    fila.put(tarefa, timeout=2)
except queue.Full:
    salvar_para_reprocessamento(tarefa)

O timeout impede espera indefinida e cria um ponto claro para métricas e fallback.

Não use empty, full ou qsize para decidir

qsize(), empty() e full() retornam estados aproximados. Outra thread pode modificar a fila imediatamente. Prefira executar put() ou get() com timeout e tratar Full ou Empty.

Encerramento com shutdown

Desde o Python 3.13, Queue.shutdown() permite encerrar produtores e consumidores de forma explícita:

import queue

fila.shutdown(immediate=False)

try:
    fila.put(nova_tarefa)
except queue.ShutDown:
    print("a fila não aceita mais tarefas")

Com immediate=False, consumidores ainda podem drenar os itens existentes e join() mantém sua garantia normal. Com immediate=True, a fila é esvaziada e join() pode retornar sem que o trabalho tenha sido executado. Use o modo imediato apenas em cancelamentos de emergência.

Consumidor compatível com shutdown

import queue

while True:
    try:
        tarefa = fila.get()
    except queue.ShutDown:
        break
    try:
        executar(tarefa)
    finally:
        fila.task_done()

Esse padrão evita sentinelas artificiais e comunica a intenção diretamente. Se o projeto também suporta versões anteriores, mantenha uma estratégia de sentinela documentada.

Sentinelas em versões antigas

PARAR = object()

def consumidor():
    while True:
        item = fila.get()
        try:
            if item is PARAR:
                return
            executar(item)
        finally:
            fila.task_done()

for _ in threads:
    fila.put(PARAR)

Envie uma sentinela por consumidor. Não reutilize um valor válido como None se ele puder ser uma tarefa real.

LifoQueue

LifoQueue funciona como pilha: o item mais recente sai primeiro. Ela pode ser útil em busca em profundidade, reutilização de recursos recentes ou processamento no qual as tarefas novas perdem valor rapidamente.

from queue import LifoQueue

fila = LifoQueue()
fila.put("antiga")
fila.put("recente")
assert fila.get() == "recente"

Uma carga contínua pode deixar itens antigos esperando. Defina política de expiração ou use FIFO quando justiça temporal importar.

PriorityQueue

A fila de prioridade entrega o menor valor primeiro:

from queue import PriorityQueue

fila = PriorityQueue()
fila.put((10, "relatório comum"))
fila.put((1, "incidente crítico"))

prioridade, tarefa = fila.get()

Quando duas prioridades são iguais, o segundo elemento também é comparado. Objetos não comparáveis causam TypeError. Use um contador de desempate ou uma dataclass ordenável:

from dataclasses import dataclass, field
from typing import Any

@dataclass(order=True)
class ItemPrioritario:
    prioridade: int
    sequencia: int
    item: Any = field(compare=False)

SimpleQueue

SimpleQueue é FIFO ilimitada e não possui task_done(), join(), maxsize nem shutdown. Ela é apropriada quando você só precisa transportar itens e controla o volume por outro mecanismo.

No CPython, sua implementação de put() é reentrante, o que permite uso em destrutores e callbacks de weakref. Para compreender esse cenário, veja o guia de weakref no Python.

Queue não é multiprocessing.Queue

queue.Queue sincroniza threads do mesmo processo. Para processos separados, use as filas de multiprocessing ou outro sistema IPC. Objetos compartilhados entre processos precisam ser serializados e possuem custos diferentes.

Tratamento de erros

Se uma tarefa falhar, decida explicitamente entre descartar, tentar novamente, mover para uma dead-letter queue ou encerrar o worker. Não faça retry infinito sem atraso e teto.

from dataclasses import replace

if tarefa.tentativas < 3:
    fila.put(replace(tarefa, tentativas=tarefa.tentativas + 1))
else:
    fila_falhas.put(tarefa)

Resultados e correlação

Uma fila de entrada e outra de saída podem transportar resultados. Inclua um ID de correlação para associar resposta e tarefa. Não espere uma resposta enquanto segura locks que o consumidor possa precisar.

Observabilidade

Registre tempo de espera, duração de processamento, taxa de erro, retries e quantidade aproximada. Use qsize() apenas como métrica, não como garantia. Para analisar gargalos dos workers, veja pstats no Python e trace no Python.

Evite deadlocks

• Não esqueça task_done().
• Não chame join() antes de permitir que consumidores trabalhem.
• Não segure locks externos durante put() bloqueante.
• Use timeout em operações que dependem de serviços externos.
• Tenha uma estratégia de shutdown.
• Não faça consumidor aguardar um item que ele próprio precisa produzir.

Testes recomendados

Teste fila cheia, fila vazia, timeout, falha no worker, shutdown normal, shutdown imediato, sentinelas, prioridades iguais e cancelamento. Use eventos para coordenar testes, evitando sleep() arbitrário.

Boas práticas

• Use fila limitada para backpressure.
• Coloque task_done() em finally.
• Trate Empty, Full e ShutDown.
• Não baseie lógica em empty().
• Documente política de retry.
• Inclua IDs de correlação.
• Meça espera e processamento.
• Escolha FIFO, LIFO ou prioridade conscientemente.

Conclusão

O queue no Python simplifica a comunicação entre threads e fornece backpressure, rastreamento de conclusão e encerramento coordenado. O resultado é um pipeline mais previsível do que listas compartilhadas e locks implementados manualmente.

Consulte a documentação oficial do queue e a documentação de threading. Para parametrizar número de workers e limites, veja configparser no Python.