asyncio.Queue.shutdown oferece uma forma explícita de encerrar filas assíncronas, desbloquear produtores e consumidores e evitar tarefas penduradas no final de uma aplicação. Em sistemas com workers, crawlers, integrações, pipelines e processamento em lote, o encerramento costuma ser mais difícil do que a execução normal. Este guia mostra como estruturar um shutdown previsível, como tratar QueueShutDown, quando usar o modo imediato e quais testes ajudam a evitar perda silenciosa de trabalho.
Por que filas assíncronas precisam de encerramento
Uma fila conecta produtores, que chamam put(), a consumidores, que chamam get(). Enquanto a aplicação está ativa, esse contrato é simples. O problema aparece quando o serviço precisa terminar. Um consumidor pode estar bloqueado esperando um item que nunca chegará. Um produtor pode estar esperando espaço em uma fila cheia. A função principal pode chamar join() e aguardar indefinidamente porque algum item não recebeu task_done().
Antes de existir uma operação específica de shutdown, era comum usar sentinelas, como None, para avisar cada worker. Esse padrão continua válido em versões antigas, mas exige disciplina: é preciso inserir uma sentinela por consumidor, impedir que valores legítimos sejam confundidos com a sentinela e decidir o que acontece com produtores bloqueados. O método shutdown() concentra esse estado na própria fila.
Modelo mental de asyncio.Queue.shutdown
Ao chamar queue.shutdown(), a fila deixa de aceitar novos itens. Chamadas futuras de put() falham com QueueShutDown, e produtores que estavam bloqueados também são liberados com a mesma exceção. Os consumidores ainda podem retirar itens já enfileirados. Quando a fila fica vazia, novas chamadas de get() levantam QueueShutDown.
Esse comportamento permite um encerramento gracioso: primeiro você impede a entrada de trabalho novo, depois deixa os workers consumirem o restante, aguarda join() e finalmente encerra as tarefas. A documentação oficial de asyncio queues deve ser a referência principal para detalhes da versão instalada. Também vale revisar a documentação de tarefas e cancelamento.
Exemplo de encerramento gracioso
import asyncio
async def worker(name: str, queue: asyncio.Queue[int]) -> None:
while True:
try:
item = await queue.get()
except asyncio.QueueShutDown:
print(f"{name}: fila encerrada")
return
try:
await asyncio.sleep(0.1)
print(f"{name}: processou {item}")
finally:
queue.task_done()
async def main() -> None:
queue: asyncio.Queue[int] = asyncio.Queue(maxsize=10)
workers = [
asyncio.create_task(worker(f"worker-{i}", queue))
for i in range(3)
]
for item in range(20):
await queue.put(item)
queue.shutdown()
await queue.join()
await asyncio.gather(*workers)
asyncio.run(main())A ordem importa. O código termina de produzir, chama shutdown(), espera todos os itens receberem task_done() e então aguarda os workers. Cada worker sai ao tentar obter um novo item depois que a fila foi drenada.
O papel de task_done e join
Para cada item retornado por get(), deve existir exatamente uma chamada de task_done(). Colocar essa chamada em finally evita que uma exceção no processamento deixe o contador interno inconsistente. Se task_done() for esquecido, join() não termina. Se for chamado mais vezes do que get(), a fila levanta erro.
O shutdown não substitui o rastreamento de tarefas. Ele apenas muda se novos itens podem entrar e como operações bloqueadas são liberadas. Em um pipeline robusto, shutdown(), join() e task_done() trabalham juntos.
Quando usar shutdown imediato
queue.shutdown(immediate=True) é apropriado quando a prioridade é parar agora, não concluir tudo. Nesse modo, a fila é drenada e operações bloqueadas são liberadas. Isso pode romper a expectativa normal de join(), pois o método pode retornar mesmo quando itens não foram realmente processados.
Use o modo imediato em falhas fatais, encerramento forçado, perda de conectividade que invalida todos os itens ou quando o processo será finalizado de qualquer maneira. Não use como atalho padrão. Em processamento financeiro, envio de mensagens, importações ou gravações, ele pode descartar trabalho. Registre métricas de itens abandonados e, quando necessário, persista a fila em um armazenamento externo.
Produtores bloqueados e backpressure
Uma fila com maxsize aplica backpressure. Quando está cheia, put() aguarda espaço. Durante o shutdown, esses produtores não devem continuar presos. A exceção QueueShutDown permite que cada produtor finalize recursos e propague um estado claro.
async def producer(queue: asyncio.Queue[str]) -> None:
for value in fonte_de_dados():
try:
await queue.put(value)
except asyncio.QueueShutDown:
salvar_checkpoint(value)
returnNão capture Exception e continue o loop, pois isso transforma o encerramento em uma repetição de falhas. Trate QueueShutDown separadamente e encerre o produtor.
Shutdown com TaskGroup
asyncio.TaskGroup combina bem com filas porque organiza a vida útil dos workers. Um coordenador pode produzir itens, iniciar o shutdown e aguardar a drenagem dentro do mesmo escopo estruturado. Para conhecer esse padrão, veja o artigo da Academify sobre TaskGroup e concorrência estruturada.
Outros conteúdos úteis são asyncio.Runner no Python, asyncio.Barrier no Python, queue.SimpleQueue no Python e asyncio.eager_task_factory. Eles ajudam a entender ciclo de vida, sincronização e custos de agendamento.
Cancelamento não é shutdown
Cancelar todos os workers pode ser necessário, mas não equivale a fechar a fila. O cancelamento interrompe tarefas; o shutdown altera o contrato da fila. Em um encerramento gracioso, prefira impedir novos itens, drenar o trabalho e só cancelar tarefas que excederem um timeout.
queue.shutdown()
try:
async with asyncio.timeout(30):
await queue.join()
except TimeoutError:
queue.shutdown(immediate=True)
for task in workers:
task.cancel()Esse padrão oferece uma janela de conclusão e depois aplica uma política de emergência. Ao capturar CancelledError, execute limpeza e propague a exceção; não a esconda sem motivo.
Erros comuns
O primeiro erro é chamar shutdown antes de todos os produtores terem terminado, sem preparar esses produtores para QueueShutDown. O segundo é esquecer task_done(). O terceiro é usar immediate=True e assumir que todos os itens foram concluídos. O quarto é criar workers sem guardar as tarefas, tornando impossível aguardá-los ou cancelá-los. O quinto é misturar sentinelas e shutdown sem uma regra clara, produzindo caminhos de saída duplicados.
Como testar
Teste uma fila vazia, uma fila com itens, uma fila cheia com produtor bloqueado e um consumidor bloqueado em get(). Verifique que novos put() falham depois do shutdown. Confirme que o modo gracioso processa todos os itens e que o imediato termina sem travar. Use timeouts nos testes para detectar deadlocks rapidamente.
Também injete falhas no worker. Mesmo quando o processamento levanta exceção, o item deve receber task_done() se foi retirado da fila. Decida se ele será reenfileirado, enviado para uma dead-letter queue ou registrado para retry.
Compatibilidade entre versões
O método e a exceção dependem da versão do Python. Bibliotecas que suportam versões anteriores podem encapsular a estratégia em um adaptador: usar shutdown() quando disponível e sentinelas quando não estiver. Evite acessar o método sem verificar a versão mínima declarada no projeto.
Conclusão
asyncio.Queue.shutdown transforma o encerramento de pipelines assíncronos em um estado explícito. O caminho seguro é interromper a entrada, drenar itens, manter a contagem de tarefas correta, aguardar workers e reservar o modo imediato para falhas reais. Com testes de produtores e consumidores bloqueados, timeouts e tratamento específico de QueueShutDown, a aplicação termina sem deadlocks e sem esconder trabalho perdido.
