Threading em Python: threads, Lock, Queue e ThreadPoolExecutor

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, multiprocessing, HTTPX com timeouts e retries e o modo free-threaded do Python 3.13.

Quando threading vale a pena

A primeira decisão não é “quantas threads usar?”, mas qual é o gargalo.

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.

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.

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.

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.

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 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.

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.

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.

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 e tratamento de erros.

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.

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.

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?

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:

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.
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; se a aplicação já for assíncrona, aprofunde-se em async/await. Escolher a ferramenta pelo tipo de trabalho evita complexidade sem ganho de performance.