O módulo concurrent.futures oferece uma interface de alto nível para executar funções concorrentemente. Em vez de criar e coordenar threads ou processos manualmente, o programa envia tarefas a um executor e recebe objetos Future que representam resultados ainda não concluídos. A mesma estrutura serve para trabalho com threads, processos e, em versões que suportam o recurso, outros tipos de executor.
A API simplifica filas, coleta de resultados, propagação de exceções, timeouts, cancelamento e encerramento. Ela não elimina os limites de concorrência: é necessário escolher o executor correto, limitar o volume de tarefas, evitar deadlocks e definir como o programa reage a falhas.
Executor e Future
Um Executor aceita chamadas. submit() retorna um Future. Esse objeto permite consultar estado, esperar, obter o resultado ou cancelar uma tarefa que ainda não começou.
from concurrent.futures import ThreadPoolExecutor
def quadrado(numero):
return numero * numero
with ThreadPoolExecutor(max_workers=4) as executor:
futuro = executor.submit(quadrado, 12)
print(futuro.result())
O bloco with chama shutdown() ao final e espera pelas tarefas conforme a política padrão. Essa forma reduz vazamentos de workers.
ThreadPoolExecutor
ThreadPoolExecutor usa threads no mesmo processo. Ele costuma ser adequado para operações que passam grande parte do tempo esperando I/O: requisições de rede, arquivos, bancos, subprocessos e APIs externas.
Threads compartilham memória, o que facilita passar objetos, mas também cria risco de corrida. Estruturas mutáveis compartilhadas precisam de locks, filas ou desenho imutável.
Trabalho limitado pela CPU
Para funções Python que consomem CPU continuamente, várias threads normalmente não oferecem paralelismo pleno devido ao lock global do interpretador em builds tradicionais. Nesses casos, processos podem ser mais apropriados.
Extensões nativas que liberam o lock podem se beneficiar de threads. Meça o workload real em vez de aplicar uma regra absoluta.
ProcessPoolExecutor
ProcessPoolExecutor usa processos separados, permitindo paralelismo em múltiplos núcleos. Argumentos e resultados precisam ser serializáveis, e a função enviada deve ser importável pelo processo filho.
from concurrent.futures import ProcessPoolExecutor
def calcular(numero):
return sum(i * i for i in range(numero))
if __name__ == "__main__":
with ProcessPoolExecutor() as executor:
futuros = [executor.submit(calcular, n) for n in (10000, 20000, 30000)]
print([f.result() for f in futuros])
A proteção if __name__ == "__main__" é essencial em plataformas que iniciam workers importando novamente o módulo principal.
Escolha o número de workers
Mais workers não significam mais velocidade. Threads demais aumentam troca de contexto, conexões simultâneas e pressão em serviços externos. Processos demais aumentam memória, serialização e competição por CPU.
Defina um limite com base no recurso mais restrito: núcleos, arquivos abertos, pool de banco, rate limit, largura de banda ou memória.
Envie tarefas com submit
submit(funcao, *args, **kwargs) oferece controle individual sobre cada tarefa.
with ThreadPoolExecutor(max_workers=8) as executor:
futuros = {
executor.submit(baixar, url, timeout=10): url
for url in urls
}
Associar o futuro à entrada ajuda a produzir diagnósticos quando uma tarefa falha.
Use map para transformações
executor.map() aplica uma função a vários elementos e retorna resultados na ordem das entradas.
with ThreadPoolExecutor(max_workers=4) as executor:
resultados = list(executor.map(processar, itens))
Essa ordem é conveniente, mas uma tarefa lenta pode atrasar a entrega dos resultados seguintes. Quando a ordem de conclusão importa, use as_completed().
Consuma com as_completed
as_completed() produz futuros à medida que terminam.
from concurrent.futures import as_completed
for futuro in as_completed(futuros):
origem = futuros[futuro]
try:
resultado = futuro.result()
except Exception as erro:
registrar_falha(origem, erro)
else:
salvar(resultado)
A exceção da função é levantada por result() na thread que coleta o futuro, preservando o fluxo de erro.
Espere grupos de tarefas
wait() permite aguardar todas, a primeira concluída ou a primeira exceção. Ele retorna conjuntos de futuros concluídos e pendentes.
from concurrent.futures import wait, FIRST_EXCEPTION
concluidos, pendentes = wait(futuros, return_when=FIRST_EXCEPTION)
Depois de detectar uma falha, decida explicitamente se as tarefas pendentes devem continuar, ser canceladas ou concluir para manter consistência.
Timeouts
future.result(timeout=...), wait() e as_completed() aceitam limites de espera. Um timeout da espera não interrompe automaticamente a função em execução.
from concurrent.futures import TimeoutError
try:
valor = futuro.result(timeout=5)
except TimeoutError:
registrar_aviso("a tarefa continua além do limite")
A própria tarefa precisa usar timeouts em rede, banco e subprocessos para poder terminar.
Cancelamento
cancel() funciona apenas quando a tarefa ainda não começou. cancelled() informa se o cancelamento ocorreu; running() indica execução em andamento.
Para cancelar trabalho já iniciado, a função deve verificar um Event, deadline ou outro sinal cooperativo.
def processar(itens, parar):
for item in itens:
if parar.is_set():
return None
executar_etapa(item)
Shutdown
shutdown(wait=True) encerra o executor e, por padrão, espera as tarefas. A opção de cancelar futuros pendentes pode ser útil durante falha ou desligamento.
Não abandone um executor global sem política. Defina quem o cria, quem o fecha e como o programa trata tarefas ainda em execução.
Deadlocks em pools de threads
Um deadlock pode ocorrer quando uma tarefa espera o resultado de outra tarefa do mesmo pool e todos os workers já estão ocupados.
def tarefa_a():
return futuro_b.result()
Evite dependências bloqueantes entre tarefas do mesmo executor. Modele a sequência no coordenador, use callbacks ou combine resultados fora do pool.
Pool com um único worker
Com apenas um worker, uma tarefa que envia outra ao mesmo executor e chama result() nunca libera o worker para a segunda tarefa. Esse padrão é uma causa clássica de travamento.
Não “corrija” apenas aumentando o pool; remova a dependência circular.
Backpressure
Enviar milhões de tarefas de uma vez consome memória para argumentos, futuros e filas. Produza trabalho em lotes ou mantenha uma janela limitada de tarefas pendentes.
def executar_em_lotes(executor, itens, tamanho=100):
lote = []
for item in itens:
lote.append(executor.submit(processar, item))
if len(lote) == tamanho:
for futuro in lote:
yield futuro.result()
lote.clear()
Uma estratégia de janela deslizante permite substituir cada futuro concluído por uma nova tarefa.
Exceções de inicialização
Executores podem receber um inicializador para configurar cada worker. Se ele falhar, o pool pode ficar inutilizável e futuros pendentes recebem erros relacionados ao executor quebrado.
Mantenha inicialização pequena, determinística e sem depender de recursos frágeis.
Estado global e processos
Processos não compartilham objetos Python comuns. Cada worker possui sua própria cópia de módulos e estado. Alterações em globals não retornam automaticamente ao processo principal.
Passe dados explicitamente e devolva resultados. Para comunicação mais complexa, consulte o artigo de posix no Python para entender processos Unix e use abstrações de multiprocessing.
Serialização
Funções locais, lambdas, generators abertos, locks e conexões geralmente não podem ser enviados a processos. Defina funções no nível do módulo e passe dados simples.
Objetos grandes aumentam custo de cópia. Às vezes é melhor cada worker abrir sua própria fonte a partir de um identificador.
Callbacks
add_done_callback() registra uma função chamada quando o futuro termina.
def ao_terminar(futuro):
try:
metricas.sucesso()
futuro.result()
except Exception:
metricas.falha()
futuro.add_done_callback(ao_terminar)
Callbacks devem ser curtos e não bloquear. Para lógica maior, envie um evento a uma fila.
Contexto e logging
Threads compartilham o logging do processo, mas contexto por requisição pode não ser propagado automaticamente. Processos precisam configurar handlers próprios ou enviar registros por fila.
Inclua identificadores de tarefa para relacionar entrada, tentativa, duração e erro.
Retries
O executor não repete tarefas automaticamente. Implemente retry apenas para falhas transitórias, com limite, atraso e idempotência.
Uma exceção de validação não melhora com repetição. Uma escrita parcial pode duplicar dados se a operação não tiver chave idempotente.
Segurança
Não envie funções ou dados serializados recebidos de usuários para um pool de processos. Serialização Python não deve ser tratada como formato seguro para entrada não confiável.
Limite concorrência para evitar negação de serviço contra seu próprio banco, filesystem ou API.
Testes
Teste sucesso, exceção, timeout, cancelamento antes do início, shutdown, pool quebrado, ordem diferente, lote vazio, tarefa muito lenta e interrupção do processo. Faça testes com poucos workers para expor dependências ocultas.
Evite assertions baseadas em timing exato. Sincronize com eventos e condições observáveis.
Erros comuns
Os erros mais frequentes são usar threads para CPU sem medir, criar processos sem a proteção do módulo principal, chamar result() dentro do mesmo pool, enviar tarefas ilimitadas, achar que timeout mata a função, ignorar exceções, passar objetos não serializáveis e esquecer de fechar o executor.
Conclusão
concurrent.futures cria uma camada uniforme para concorrência com tarefas. Use threads para I/O, processos para CPU quando fizer sentido, limite workers, aplique backpressure e trate cada futuro como uma operação que pode falhar ou ser cancelada.
Consulte a documentação oficial de concurrent.futures e o guia de signal no Python para coordenar encerramento.
