O módulo queue oferece filas sincronizadas para trocar trabalho entre threads. Ele evita que você implemente travas ao redor de uma lista e torna explícito quando o produtor deve esperar porque os consumidores estão ocupados.

from queue import Queue
from threading import Thread

fila: Queue[int | None] = Queue(maxsize=100)

def consumidor() -> None:
    while True:
        item = fila.get()
        try:
            if item is None:
                return
            print(item * 2)
        finally:
            fila.task_done()

worker = Thread(target=consumidor)
worker.start()
for valor in range(5):
    fila.put(valor)
fila.put(None)
fila.join()
worker.join()

put() espera quando a fila limitada está cheia, enquanto get() espera por um item. Cada retirada precisa de um task_done(), inclusive quando o processamento falha, por isso o bloco finally é importante. join() aguarda o contador de tarefas pendentes chegar a zero.

Boas práticas

Use Queue para FIFO, LifoQueue para pilha e PriorityQueue para itens ordenados por prioridade. Não use qsize() para decidir uma ação concorrente: o tamanho pode mudar antes da próxima instrução. Defina também uma estratégia de encerramento, como um sentinela por consumidor ou a API de shutdown disponível na versão adotada.

Para escolher entre threads e processos, consulte multithreading e multiprocessing em Python. Se o fluxo roda em asyncio, use a fila específica apresentada em async e await.

A documentação oficial do módulo queue, consultada em 22 de julho de 2026, detalha a API e suas garantias.

Por que uma fila é melhor que uma lista compartilhada

Uma lista protegida por Lock pode funcionar, mas logo exige decisões sobre espera, sinalização e encerramento. Fazer polling com if itens: desperdiça CPU ou adiciona latência. Queue reúne armazenamento, bloqueio e notificação em uma abstração testada. Suas operações são sincronizadas entre threads, portanto um produtor pode inserir enquanto consumidores retiram sem corromper a estrutura.

Isso não torna o processamento inteiro thread-safe. Se dois consumidores atualizam o mesmo dicionário, arquivo ou objeto mutável, esse recurso ainda precisa de coordenação. A fila protege a passagem de itens, não todos os efeitos que acontecem depois de get().

Backpressure com maxsize

Uma fila ilimitada permite que produtores rápidos consumam memória enquanto o destino está lento. Com Queue(maxsize=20), o vigésimo primeiro item faz put() aguardar espaço. Esse bloqueio propaga a lentidão e limita trabalho pendente. O tamanho adequado depende do custo de cada item, do pico aceitável de memória e do tempo esperado de processamento.

Também existem put_nowait() e get_nowait(). Elas lançam Full ou Empty em vez de esperar. São úteis quando a aplicação tem uma política explícita, como descartar telemetria de baixa prioridade, mas não devem ser usadas com uma checagem anterior de full() ou empty(). Entre a consulta e a operação, outra thread pode mudar a fila.

from queue import Full, Queue

eventos: Queue[str] = Queue(maxsize=2)

def publicar(evento: str) -> bool:
    try:
        eventos.put(evento, timeout=0.5)
        return True
    except Full:
        return False

O timeout evita espera infinita e permite registrar, tentar novamente ou cancelar de maneira controlada.

Vários consumidores e encerramento

Ao usar sentinelas, envie uma para cada consumidor. Uma única sentinela encerra apenas a thread que a retirar. Escolha um objeto que nunca possa ser confundido com trabalho válido. None é suficiente quando os itens reais nunca são nulos; em APIs genéricas, um object() privado é mais seguro.

from queue import Queue
from threading import Thread

PARAR = object()
trabalhos: Queue[object] = Queue()

def executar() -> None:
    while True:
        trabalho = trabalhos.get()
        try:
            if trabalho is PARAR:
                return
            processar(trabalho)
        except Exception:
            registrar_falha(trabalho)
        finally:
            trabalhos.task_done()

workers = [Thread(target=executar, name=f"worker-{i}") for i in range(3)]
for worker in workers:
    worker.start()

for item in carregar_trabalhos():
    trabalhos.put(item)
for _ in workers:
    trabalhos.put(PARAR)

trabalhos.join()
for worker in workers:
    worker.join()

join() e Thread.join() resolvem problemas diferentes. O primeiro espera todos os itens receberem task_done(); o segundo espera a thread terminar. Usar ambos confirma que o trabalho acabou e que nenhum worker ficou vivo.

Tratamento de erros e o contador de tarefas

Queue incrementa o contador interno a cada put(). task_done() decrementa esse contador, não remove um item. Chamá-lo mais vezes que get() causa ValueError; esquecê-lo faz join() bloquear para sempre. Por isso, mantenha task_done() em finally, mas somente depois de uma retirada bem-sucedida.

Decida também o destino das falhas. Uma exceção não capturada encerra o consumidor e pode deixar a fila parada. Em sistemas reais, registre contexto, envie o item para uma fila de falhas ou faça repetição com limite. Recolocar indefinidamente o mesmo item cria um loop e pode impedir o encerramento.

FIFO, pilha e prioridade

Queue entrega itens em ordem aproximada de chegada e é a escolha padrão. LifoQueue entrega o item mais recente primeiro, o que pode ser útil para explorar uma árvore, mas tarefas antigas podem ficar esperando. PriorityQueue retira o menor valor primeiro. Uma tupla (prioridade, item) funciona apenas se os itens puderem ser comparados quando prioridades empatam.

from dataclasses import dataclass, field
from queue import PriorityQueue
from typing import Any

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

fila_prioritaria: PriorityQueue[Trabalho] = PriorityQueue()
fila_prioritaria.put(Trabalho(2, 0, {"tipo": "relatorio"}))
fila_prioritaria.put(Trabalho(1, 1, {"tipo": "alerta"}))

O número sequencial desempata sem comparar dicionários. Defina claramente se prioridade menor significa maior urgência.

Queue, SimpleQueue, asyncio e multiprocessing

SimpleQueue oferece uma FIFO não limitada com API menor. Ela serve quando não há necessidade de maxsize, task_done() ou join(). asyncio.Queue coordena corrotinas no mesmo event loop e deve ser aguardada com await; não é substituta para comunicação arbitrária entre threads. Já multiprocessing.Queue serializa dados para atravessar processos e tem custos e regras diferentes.

Threads costumam ajudar em trabalho de I/O, como muitas requisições ou arquivos. Para cálculo Python intensivo, o GIL pode limitar ganho e processos podem ser mais adequados. A fila organiza o fluxo, mas não escolhe o modelo de concorrência pela aplicação.

Testes e observabilidade

Teste com filas pequenas para forçar backpressure e use timeouts para que uma falha produza um erro claro, não uma suíte travada. Verifique caminho feliz, exceção no consumidor, fila cheia e encerramento com vários workers. Evite afirmar uma ordem entre consumidores: o escalonamento das threads não é determinístico.

Em produção, acompanhe itens processados, falhas, duração e tempo de espera. qsize() pode ser uma métrica aproximada, embora não seja base segura para decisões de sincronização. Um crescimento contínuo indica que a taxa de entrada supera a capacidade dos consumidores. A resposta pode ser reduzir produção, aumentar workers dentro dos limites do destino ou otimizar o processamento.

Checklist de implementação

Escolha uma capacidade finita quando o produtor puder superar o consumidor. Garanta um task_done() para cada get(). Planeje cancelamento e encerramento antes de iniciar as threads. Trate exceções dentro do worker e preserve contexto suficiente para diagnóstico. Por fim, documente se a entrega pode ser repetida ou perdida: Queue coordena memória local, mas não oferece persistência, transação nem recuperação após o processo terminar.