--- title: "Threading em Python: threads, Lock, Queue e ThreadPoolExecutor" url: "https://python.dev.br/blog/python-threading-threadpoolexecutor/" markdown_url: "https://python.dev.br/blog/python-threading-threadpoolexecutor.MD" description: "Guia prático de threading em Python para tarefas de I/O: ThreadPoolExecutor, Thread, Lock, Queue, Event, Semaphore, timeouts, erros e boas práticas." date: "2026-07-23" author: "Equipe Python Brasil" --- # Threading em Python: threads, Lock, Queue e ThreadPoolExecutor Guia prático de threading em Python para tarefas de I/O: ThreadPoolExecutor, Thread, Lock, Queue, Event, Semaphore, timeouts, erros e boas práticas. **Para tarefas que passam boa parte do tempo esperando**, como consultar APIs, ler muitos arquivos, verificar URLs ou conversar com um banco de dados, `threading` pode reduzir bastante o tempo total sem trocar toda a aplicação por `asyncio`. A recomendação prática é direta: comece com `ThreadPoolExecutor` para lotes independentes; use `threading.Thread` quando precisar de workers persistentes; compartilhe o mínimo de estado possível; e adicione `Lock`, `Queue`, `Event` ou `Semaphore` somente quando houver uma necessidade concreta de coordenação. Este guia explica threads no CPython tradicional, o efeito do GIL, tratamento de erros, cancelamento cooperativo e um exemplo de automação útil para o contexto brasileiro. Ele complementa os artigos sobre [async/await](/blog/python-async-await/), [multiprocessing](/blog/python-multiprocessing/), [HTTPX com timeouts e retries](/blog/python-httpx-requests-moderno/) e o modo [free-threaded do Python 3.13](/blog/python-3-13-free-threaded-sem-gil/). ## Quando threading vale a pena A primeira decisão não é “quantas threads usar?”, mas **qual é o gargalo**. | Tipo de tarefa | Exemplos | Ferramenta inicial | |---|---|---| | I/O-bound | HTTP, banco, arquivos, sockets, espera de subprocesso | `threading` ou `asyncio` | | CPU-bound | compressão pesada, cálculo puro, transformação de imagem em Python | `multiprocessing` / `ProcessPoolExecutor` | | Mista | baixar arquivos e depois processá-los | threads no download, processos no cálculo | | Worker persistente | consumir uma fila até receber sinal de parada | `Thread` + `Queue` + `Event` | No CPython padrão, o **Global Interpreter Lock (GIL)** impede que duas threads executem bytecode Python simultaneamente no mesmo processo. Isso não torna threads inúteis: quando uma thread espera rede, disco ou outra operação que libera o GIL, outra pode avançar. Por isso, dez requisições HTTP com latência de 500 ms podem terminar muito antes quando sobrepostas. Já quatro threads executando um loop matemático escrito em Python provavelmente não usarão quatro núcleos de forma eficiente. Para esse segundo caso, veja o guia de [multiprocessing em Python](/blog/python-multiprocessing/). > O build free-threaded introduzido experimentalmente no Python 3.13 muda parte dessa análise, mas não é o modo padrão de todas as instalações nem elimina a necessidade de sincronização. Código correto não deve presumir que o GIL protege automaticamente o estado compartilhado. ## A opção moderna: ThreadPoolExecutor `ThreadPoolExecutor`, do módulo `concurrent.futures`, gerencia um conjunto de threads e representa cada execução com um `Future`. Para tarefas independentes, costuma ser mais legível e seguro do que criar threads manualmente. ```python from concurrent.futures import ThreadPoolExecutor from time import sleep def consultar_pedido(pedido_id: int) -> dict: # Simula uma API ou consulta de banco bloqueante sleep(0.3) return {"id": pedido_id, "status": "processado"} pedidos = [101, 102, 103, 104, 105] with ThreadPoolExecutor(max_workers=4) as executor: resultados = list(executor.map(consultar_pedido, pedidos)) print(resultados) ``` `executor.map` preserva a ordem das entradas. Isso é conveniente para relatórios, mas pode atrasar o consumo se a primeira tarefa for lenta. Quando você quer tratar cada resultado assim que ele termina, use `submit` e `as_completed`. ```python from concurrent.futures import ThreadPoolExecutor, as_completed pedidos = [101, 102, 103, 104, 105] with ThreadPoolExecutor(max_workers=4) as executor: futuros = { executor.submit(consultar_pedido, pedido_id): pedido_id for pedido_id in pedidos } for futuro in as_completed(futuros): pedido_id = futuros[futuro] try: resultado = futuro.result() except Exception as exc: print(f"Pedido {pedido_id} falhou: {exc}") else: print(f"Pedido {pedido_id}: {resultado['status']}") ``` A chamada a `future.result()` é importante: é nela que uma exceção levantada dentro da thread volta para a thread principal. Submeter tarefas e nunca inspecionar os futures é um erro comum, pois falhas podem passar despercebidas. ## Timeouts e tarefas travadas Concorrência sem limite de espera transforma uma lentidão externa em fila acumulada. Defina timeout no cliente de I/O **e**, quando fizer sentido, também ao aguardar o `Future`. ```python from concurrent.futures import ThreadPoolExecutor, TimeoutError from time import sleep def operacao_lenta() -> str: sleep(5) return "ok" with ThreadPoolExecutor(max_workers=1) as executor: futuro = executor.submit(operacao_lenta) try: print(futuro.result(timeout=1)) except TimeoutError: print("A operação ultrapassou o tempo de espera") ``` Um detalhe essencial: o timeout de `result()` limita **quanto a thread chamadora espera**, mas não interrompe à força uma função que já começou. `future.cancel()` só cancela uma tarefa que ainda não está executando. A função trabalhadora precisa ter timeouts próprios ou consultar um sinal de cancelamento cooperativo. Em chamadas HTTP, configure os limites na biblioteca cliente. O artigo sobre [HTTPX em Python](/blog/python-httpx-requests-moderno/) mostra como separar timeout de conexão, leitura, escrita e pool. ## Quantas threads usar? Não existe número universal. O valor depende da latência, dos limites do serviço, da memória e da capacidade do destino. Uma estratégia segura: 1. comece com 4 a 8 workers; 2. meça duração, erros e uso de recursos; 3. verifique rate limits da API e limite de conexões do banco; 4. aumente gradualmente, se houver ganho real; 5. pare quando a latência ou a taxa de erro começar a subir. Abrir 100 threads não transforma automaticamente uma automação em algo rápido. Pode apenas criar 100 conexões concorrentes, consumir descritores de arquivo, sobrecarregar o banco ou receber HTTP 429. Para controlar a pressão sobre um recurso específico, use `Semaphore`. ```python import threading limite_api = threading.Semaphore(3) def consultar_com_limite(item_id: int) -> dict: with limite_api: return consultar_pedido(item_id) ``` Mesmo que o pool tenha 12 workers, no máximo três entram simultaneamente no bloco protegido. ## Criando uma Thread diretamente Use `threading.Thread` quando a tarefa tem ciclo de vida próprio: um monitor, consumidor de fila, atualizador de cache ou worker que fica ativo durante a aplicação. ```python import threading from time import sleep def monitorar(nome: str, parar: threading.Event) -> None: while not parar.is_set(): print(f"{nome}: verificando fila") # wait permite acordar imediatamente quando o Event for sinalizado parar.wait(timeout=2) print(f"{nome}: encerrado") parar = threading.Event() worker = threading.Thread( target=monitorar, args=("monitor-nfe", parar), name="monitor-nfe", ) worker.start() sleep(5) parar.set() worker.join(timeout=3) ``` `join()` espera a thread terminar. Sem ele, o fluxo principal pode encerrar recursos enquanto o worker ainda os usa. Dar um `name` descritivo também melhora logs e diagnósticos. ### Daemon thread exige cuidado Uma thread com `daemon=True` não impede o processo de encerrar. Isso serve para tarefas auxiliares descartáveis, mas é perigoso se a thread grava arquivo, confirma mensagem ou mantém uma transação: o processo pode terminar sem limpeza. Para trabalho importante, prefira thread não-daemon e desligamento cooperativo com `Event` + `join`. ## Queue: comunicação segura entre threads `queue.Queue` é o padrão para produtor–consumidor. Ela já implementa sincronização, espera quando está vazia e pode aplicar capacidade máxima para criar **backpressure**. ```python import threading from queue import Queue fila: Queue[str | None] = Queue(maxsize=100) def consumidor() -> None: while True: caminho = fila.get() try: if caminho is None: return print(f"Processando {caminho}") # validar XML, extrair dados, salvar resultado... finally: fila.task_done() workers = [ threading.Thread(target=consumidor, name=f"xml-worker-{i}") for i in range(3) ] for worker in workers: worker.start() for caminho in ["nfe-001.xml", "nfe-002.xml", "nfe-003.xml"]: fila.put(caminho) # Um sentinela por consumidor for _ in workers: fila.put(None) fila.join() for worker in workers: worker.join() ``` O sentinela `None` comunica que não haverá novos itens. O `finally` garante `task_done()` mesmo quando o processamento falha. Sem esse cuidado, `fila.join()` pode esperar para sempre. Para automações com documentos fiscais, combine a fila com as práticas de [leitura de XML de NF-e](/blog/python-automacao-notas-fiscais-xml-nfe/) e [tratamento de erros](/blog/tratamento-de-erros-python/). ## Lock e a condição de corrida Uma condição de corrida ocorre quando o resultado depende da ordem imprevisível das threads. Imagine várias threads atualizando um contador, um saldo acumulado ou o mesmo dicionário. ```python import threading contador = 0 lock = threading.Lock() def incrementar(vezes: int) -> None: global contador for _ in range(vezes): with lock: contador += 1 threads = [threading.Thread(target=incrementar, args=(10_000,)) for _ in range(4)] for thread in threads: thread.start() for thread in threads: thread.join() print(contador) # 40000 ``` Use `with lock:` em vez de `lock.acquire()` e `lock.release()` manuais. O context manager libera o lock mesmo se ocorrer exceção. Mas não trate `Lock` como solução automática. A melhor arquitetura geralmente é: - cada worker recebe uma entrada imutável; - cada worker devolve seu resultado; - a thread principal agrega os resultados; - estado compartilhado fica restrito a uma pequena região. Isso reduz deadlocks e facilita testes. Se uma função adquire o mesmo lock recursivamente, talvez seja necessário `RLock`, mas primeiro questione o desenho: locks reentrantes podem esconder acoplamento excessivo. ## Exemplo prático: verificar endpoints em paralelo O exemplo abaixo mede vários endpoints sem compartilhar uma lista mutável entre workers. Cada tarefa devolve um objeto, e a thread principal monta o relatório. ```python from concurrent.futures import ThreadPoolExecutor, as_completed from dataclasses import dataclass from time import perf_counter from urllib.error import HTTPError, URLError from urllib.request import Request, urlopen @dataclass(frozen=True) class Resultado: url: str status: int | None duracao_ms: int erro: str | None = None def verificar(url: str) -> Resultado: inicio = perf_counter() request = Request(url, headers={"User-Agent": "monitor-interno/1.0"}) try: with urlopen(request, timeout=5) as response: status = response.status erro = None except HTTPError as exc: status = exc.code erro = f"HTTP {exc.code}" except (URLError, TimeoutError) as exc: status = None erro = str(exc.reason) if isinstance(exc, URLError) else "timeout" duracao_ms = round((perf_counter() - inicio) * 1000) return Resultado(url, status, duracao_ms, erro) urls = [ "https://python.dev.br/", "https://python.dev.br/blog/", "https://python.dev.br/vagas/", ] resultados: list[Resultado] = [] with ThreadPoolExecutor(max_workers=3, thread_name_prefix="healthcheck") as executor: futuros = {executor.submit(verificar, url): url for url in urls} for futuro in as_completed(futuros): url = futuros[futuro] try: resultados.append(futuro.result()) except Exception as exc: resultados.append(Resultado(url, None, 0, repr(exc))) for item in sorted(resultados, key=lambda r: r.url): print(item) ``` Pontos importantes: - `dataclass(frozen=True)` torna o resultado imutável; - o cliente tem timeout; - cada `Future` fica associado à URL; - exceções inesperadas também entram no relatório; - `thread_name_prefix` facilita identificar workers no log; - a ordenação acontece depois, sem lock compartilhado. Em produção, respeite os termos de uso dos serviços monitorados, aplique intervalos razoáveis e não use concorrência para contornar rate limits. ## Threading, asyncio ou multiprocessing? | Critério | Threading | asyncio | Multiprocessing | |---|---|---|---| | Melhor para | I/O bloqueante | Muito I/O com APIs async | CPU-bound | | Integração com código síncrono | Excelente | Pode exigir adaptação | Boa, com serialização | | Memória compartilhada | Sim | Sim, no mesmo event loop | Não por padrão | | Risco típico | corrida e deadlock | bloquear o event loop | overhead e pickle | | API inicial | `ThreadPoolExecutor` | `asyncio.TaskGroup` | `ProcessPoolExecutor` | Escolha threading quando você já usa bibliotecas síncronas e precisa sobrepor esperas. Escolha `asyncio` quando a aplicação inteira — servidor, cliente HTTP, banco e filas — já oferece APIs assíncronas e precisa manter muitas conexões. Escolha processos quando o tempo é gasto calculando em Python. Também é possível chamar código bloqueante a partir de uma aplicação async com `asyncio.to_thread`, evitando bloquear o event loop: ```python import asyncio async def main() -> None: resultado = await asyncio.to_thread(consultar_pedido, 101) print(resultado) asyncio.run(main()) ``` Isso é uma ponte, não um motivo para colocar toda função síncrona em thread sem medir. ## Boas práticas e armadilhas Use este checklist em code review: 1. **Confirme que o gargalo é I/O-bound.** Meça antes e depois. 2. **Prefira `ThreadPoolExecutor`** para lotes de tarefas curtas e independentes. 3. **Sempre inspecione `Future.result()`.** Não esconda exceções dos workers. 4. **Defina timeouts no I/O.** Timeout apenas no Future não mata a operação. 5. **Evite estado mutável compartilhado.** Retorne resultados ou use `Queue`. 6. **Use `with lock:`** quando a seção crítica for realmente necessária. 7. **Não faça I/O enquanto segura um lock.** Uma rede lenta pode bloquear todas as threads. 8. **Limite concorrência externa.** Respeite pools de conexão e rate limits. 9. **Implemente encerramento cooperativo.** `Event`, sentinela e `join` são mais seguros que abandono abrupto. 10. **Dê nomes às threads.** Logs ficam muito mais úteis. 11. **Teste falhas, não só o caminho feliz.** Simule timeout, resposta inválida e exceção no worker com [pytest](/blog/testes-unitarios-python/). 12. **Não suponha que operações compostas sejam atômicas.** O comportamento pode mudar entre implementações e builds free-threaded. ## Conclusão `threading` continua sendo uma ferramenta valiosa do Python moderno porque grande parte das automações reais espera por rede, disco, banco ou serviços externos. Para a maioria dos casos, `ThreadPoolExecutor` entrega a melhor relação entre simplicidade e controle. `Thread`, `Queue`, `Event`, `Lock` e `Semaphore` entram quando existe um ciclo de vida ou uma coordenação mais específica. O caminho seguro é menos “criar muitas threads” e mais **desenhar tarefas isoladas, limitar concorrência, definir timeouts e tornar falhas visíveis**. Se o gargalo for cálculo, migre para [multiprocessing](/blog/python-multiprocessing/); se a aplicação já for assíncrona, aprofunde-se em [async/await](/blog/python-async-await/). Escolher a ferramenta pelo tipo de trabalho evita complexidade sem ganho de performance.