O módulo queue fornece filas sincronizadas para comunicação segura entre threads. Ele implementa FIFO, LIFO, prioridade e uma fila simples, com operações bloqueantes, timeouts, capacidade máxima e acompanhamento de tarefas. O padrão produtor-consumidor fica mais claro porque os workers trocam mensagens em vez de modificar estruturas compartilhadas diretamente.
Uma fila não torna o processamento automaticamente correto. O programa ainda precisa definir ownership, formato das mensagens, backpressure, tratamento de exceções, encerramento e idempotência. Também é importante distinguir queue.Queue, destinada a threads, de multiprocessing.Queue e asyncio.Queue.
Crie uma fila FIFO
Queue entrega itens na ordem de entrada.
from queue import Queue
fila = Queue()
fila.put("primeiro")
fila.put("segundo")
print(fila.get())
A fila sincroniza o acesso interno entre threads.
Defina capacidade
maxsize limita a quantidade aproximada de itens pendentes.
fila = Queue(maxsize=100)
Uma fila limitada cria backpressure: produtores podem esperar quando consumidores não acompanham.
put bloqueante
Por padrão, put() espera até existir espaço.
fila.put(item, timeout=5)
Com timeout, a operação gera queue.Full se não puder inserir.
put_nowait
put_nowait() tenta inserir imediatamente.
from queue import Full
try:
fila.put_nowait(item)
except Full:
registrar_descarte(item)
Defina uma política: esperar, descartar, persistir, reduzir produção ou retornar erro.
get bloqueante
get() espera por um item.
item = fila.get(timeout=2)
Se o prazo terminar, gera queue.Empty.
Não use empty para decidir get
empty(), full() e qsize() são apenas aproximações em ambiente concorrente.
Outra thread pode alterar a fila entre a consulta e a operação. Use get_nowait() ou timeout e trate a exceção.
Produtor e consumidor
from threading import Thread
from queue import Queue
fila = Queue(maxsize=20)
def consumidor():
while True:
item = fila.get()
try:
processar(item)
finally:
fila.task_done()
thread = Thread(target=consumidor, daemon=True)
thread.start()
O finally garante que o contador de tarefas seja atualizado mesmo após erro.
task_done
Cada item retirado com get() deve receber exatamente uma chamada a task_done() quando o trabalho associado terminar.
Chamar mais vezes gera ValueError; esquecer a chamada pode bloquear join() indefinidamente.
join
fila.join() espera até que todas as tarefas inseridas tenham sido marcadas como concluídas.
for item in itens:
fila.put(item)
fila.join()
Isso não encerra workers. Apenas aguarda o contador de tarefas.
Sentinelas para shutdown
Um objeto especial pode informar que o consumidor deve parar.
PARAR = object()
def consumidor():
while True:
item = fila.get()
try:
if item is PARAR:
return
processar(item)
finally:
fila.task_done()
Envie uma sentinela por consumidor quando cada thread retira independentemente.
Ordem do shutdown
Pare os produtores, aguarde a entrada normal, envie sentinelas, espere join() e depois faça join das threads.
Uma ordem incorreta pode deixar trabalho depois das sentinelas ou bloquear produtores em uma fila cheia.
Shutdown moderno
Versões recentes do Python oferecem mecanismos explícitos de shutdown para filas sincronizadas. Eles permitem impedir novas inserções e acordar operações bloqueadas conforme a política escolhida.
Verifique a versão mínima do projeto e trate a exceção específica de fila encerrada. Para compatibilidade ampla, o padrão de sentinelas continua útil.
LifoQueue
LifoQueue entrega o item mais recente primeiro.
from queue import LifoQueue
pilha = LifoQueue()
pilha.put("a")
pilha.put("b")
print(pilha.get())
É útil em buscas ou caches de trabalho recente, mas itens antigos podem sofrer starvation.
PriorityQueue
PriorityQueue entrega o menor valor primeiro e usa heapq internamente.
from queue import PriorityQueue
fila = PriorityQueue()
fila.put((10, "normal"))
fila.put((1, "urgente"))
Use prioridade, contador e payload para evitar comparação entre tarefas.
Empates em PriorityQueue
from itertools import count
contador = count()
fila.put((prioridade, next(contador), tarefa))
O contador preserva estabilidade e impede que objetos não comparáveis sejam avaliados.
SimpleQueue
SimpleQueue oferece uma FIFO não limitada com API menor.
from queue import SimpleQueue
fila = SimpleQueue()
fila.put(item)
item = fila.get()
Use quando não precisa de maxsize, task tracking ou backpressure interno.
Escolha a fila correta
Use Queue para FIFO com capacidade e acompanhamento, LifoQueue para pilha sincronizada, PriorityQueue para prioridades e SimpleQueue para comunicação simples sem limite.
Documente a escolha porque ela afeta fairness e memória.
Exceções no consumidor
Uma falha não deve matar silenciosamente todos os workers.
try:
processar(item)
except Exception as erro:
registrar_falha(item, erro)
finally:
fila.task_done()
Decida entre retry, dead-letter, parada do sistema ou continuação.
Retries
Não reinsira indefinidamente o mesmo item na fila principal.
Mantenha contagem de tentativas, backoff, deadline e uma fila de falhas. Operações com side effects precisam ser idempotentes.
Vários consumidores
Mais threads podem aumentar throughput para I/O, mas também ampliam concorrência contra banco, API e filesystem.
Dimensione workers com base no recurso downstream, não apenas na CPU.
GIL e CPU
Threads normalmente não aceleram trabalho Python puramente limitado por CPU em builds tradicionais.
Para CPU, avalie processos ou extensões que liberem o GIL. Consulte multiprocessing no Python.
Dados mutáveis
A fila transfere uma referência, não uma cópia profunda.
Depois de put(), o produtor deve tratar o objeto como pertencente ao consumidor ou enviar uma estrutura imutável.
Mensagens
Use dataclasses, NamedTuple ou objetos pequenos com campos claros: tipo, payload, ID, tentativa e deadline.
Evite tuplas posicionais longas e dicionários sem esquema.
Backpressure
Uma fila limitada impede crescimento ilimitado, mas produtores bloqueados também podem causar deadlock.
Nunca faça put() bloqueante enquanto mantém um lock que o consumidor precisa.
Fairness
A ordem de wakeup entre threads não deve ser tratada como garantia de justiça perfeita.
Se fairness for requisito de negócio, modele explicitamente classes, cotas ou filas separadas.
Integração com selectors
Workers podem enviar resultados por uma fila; o loop de I/O lê comandos após ser acordado por socketpair ou pipe.
Veja selectors no Python.
Integração com ThreadPoolExecutor
Executores já mantêm uma fila interna. Não adicione outra camada sem necessidade.
Uma fila própria é útil quando você precisa de capacidade, prioridade, mensagens persistentes no processo ou workers customizados.
Timeouts e cancelamento
Timeout em get() permite que a thread verifique uma flag de parada.
while not parar.is_set():
try:
item = fila.get(timeout=0.5)
except Empty:
continue
Cancelamento de trabalho já iniciado precisa ser cooperativo.
Daemon threads
Threads daemon não impedem o processo de terminar, mas podem ser interrompidas sem cleanup.
Para dados importantes, use threads normais e shutdown explícito.
Observabilidade
Meça tamanho aproximado, tempo de espera, idade do item, taxa de entrada, throughput, falhas e retries.
Não use qsize() como garantia lógica, mas ele pode servir como métrica aproximada.
Segurança
Limite tamanho e quantidade de mensagens. Não coloque callbacks arbitrários recebidos de usuários.
Redija segredos em logs e aplique autorização antes de produzir tarefas.
Testes
Teste fila vazia e cheia, timeout, vários produtores, vários consumidores, exceção, sentinela, retry, shutdown, task_done() e deadlock por lock.
Use eventos de sincronização em vez de sleeps exatos.
Erros comuns
Os erros mais frequentes são usar empty() como garantia, esquecer task_done(), confundir join() da fila com join da thread, enviar poucas sentinelas, permitir fila ilimitada, bloquear segurando lock e compartilhar objetos mutáveis depois de put().
Conclusão
queue oferece comunicação sincronizada e segura entre threads. Use capacidade para backpressure, task_done()/join() para acompanhamento e sentinelas ou shutdown explícito para encerramento.
Consulte a documentação oficial de queue, heapq no Python e selectors no Python.
