Concorrencia Python: Threading vs Multiprocessing vs Asyncio (quando usar cada um)

Concorrencia Python é um tema recorrente para quem constrói APIs, robôs de automação, scrapers, pipelines de dados e serviços que precisam lidar com muitas tarefas “ao mesmo tempo”. Em Python, três abordagens dominam o assunto: threading, multiprocessing e asyncio.

Apesar de parecerem equivalentes, elas resolvem problemas diferentes e têm implicações diretas em desempenho, custos, complexidade e até segurança operacional (por exemplo, controle de timeouts, isolamento de falhas e consumo de recursos). Este guia detalha, de forma prática, como escolher a técnica adequada e como evitar armadilhas comuns.

1) Conceitos essenciais: concorrência vs paralelismo

Antes de comparar ferramentas, vale separar dois termos:

Concorrência: lidar com múltiplas tarefas de forma intercalada (você progride em várias “frentes”, alternando execução). Nem sempre há execução simultânea real.
Paralelismo: executar tarefas literalmente ao mesmo tempo, em múltiplos núcleos de CPU.

Em Python, threading e asyncio são frequentemente usados para concorrência, enquanto multiprocessing é o caminho típico para paralelismo em tarefas pesadas de CPU.

2) O fator decisivo: o GIL (Global Interpreter Lock)

A implementação mais usada do Python (CPython) possui o GIL, um mecanismo que garante que apenas uma thread execute bytecode Python por vez dentro de um mesmo processo.

Impacto prático:

Em tarefas CPU-bound (cálculo pesado), threads não escalam com múltiplos núcleos em CPython; você tende a ganhar pouco (ou até perder) por overhead de agendamento.
Em tarefas I/O-bound (rede, disco, banco, APIs), threads funcionam bem, porque o tempo “parado” aguardando I/O permite que outras threads avancem.

O GIL não impede paralelismo quando você usa multiprocessing (processos distintos têm interpretadores distintos) e não costuma ser um obstáculo no asyncio para I/O, porque o modelo é diferente: não há múltiplas threads executando Python em paralelo; há uma thread principal alternando coroutines.

3) Threading: ideal para I/O concorrente (com ressalvas)

Quando faz sentido

Use threading quando você precisa de concorrência para tarefas principalmente de I/O:

chamadas HTTP para várias URLs
leitura/escrita em arquivos
operações com banco (depende do driver)
automações que ficam aguardando respostas externas

Vantagens

Modelo mental direto para quem já conhece programação sequencial.
Bom para integrar com bibliotecas bloqueantes (que não oferecem versão assíncrona).
Overhead menor que criar múltiplos processos.

Limitações e riscos

Não acelera CPU-bound em CPython por causa do GIL.
Concorrência com threads aumenta riscos de:
- race conditions
- deadlocks
- inconsistência de estado compartilhado
Depuração é mais difícil (bugs “intermitentes”).

Exemplo (I/O com ThreadPoolExecutor)

from concurrent.futures import ThreadPoolExecutor, as_completed
import requests

URLS = [
    "https://example.com",
    "https://example.org",
    "https://example.net",
]

def fetch(url: str) -> tuple[str, int]:
    r = requests.get(url, timeout=10)
    return url, r.status_code

with ThreadPoolExecutor(max_workers=20) as pool:
    futures = [pool.submit(fetch, url) for url in URLS]
    for fut in as_completed(futures):
        url, status = fut.result()
        print(url, status)

Pontos de atenção:

timeout é obrigatório para evitar threads presas indefinidamente.
max_workers deve ser calibrado (exagerar pode derrubar desempenho e saturar recursos).

4) Multiprocessing: o caminho padrão para CPU-bound

Quando faz sentido

Use multiprocessing quando há cálculo pesado e você quer usar múltiplos núcleos:

processamento de imagens
compressão/criptografia (dependendo da lib)
parsing e transformações grandes
cálculos numéricos (quando não delegados a libs que já paralelizam)
tarefas de ML que não estão totalmente em NumPy/PyTorch (que podem liberar GIL ou usar paralelismo próprio)

Vantagens

Contorna o GIL: cada processo tem seu interpretador e pode rodar em paralelo.
Isolamento maior: um crash em um processo não derruba necessariamente o mestre (depende de como você gerencia).

Custos e armadilhas

Overhead maior: criar processos é mais caro que threads.
Comunicação entre processos é mais lenta (serialização/pickle).
Em Windows e macOS (default “spawn”), é comum cair em erros se não houver:
- if __name__ == "__main__":
Objetos precisam ser “picklable” para trafegar entre processos.
Pode consumir muita memória (processos duplicam estado com mais facilidade que threads).

Exemplo (CPU-bound com ProcessPoolExecutor)

from concurrent.futures import ProcessPoolExecutor
import math

def is_prime(n: int) -> bool:
    if n < 2:
        return False
    if n % 2 == 0:
        return n == 2
    r = int(math.isqrt(n))
    for i in range(3, r + 1, 2):
        if n % i == 0:
            return False
    return True

def count_primes(limit: int) -> int:
    return sum(1 for x in range(limit) if is_prime(x))

if __name__ == "__main__":
    limits = [200_000, 200_000, 200_000, 200_000]
    with ProcessPoolExecutor() as pool:
        results = list(pool.map(count_primes, limits))
    print(results)

5) Asyncio: alta escala para I/O, com arquitetura assíncrona

Quando faz sentido

Use asyncio quando seu problema é I/O intenso e você quer lidar com muitas conexões simultâneas com eficiência:

APIs e microserviços de alta concorrência
WebSockets
crawlers e pipelines de rede
clientes que fazem milhares de requests concorrentes

A vantagem do asyncio aparece quando:

as operações são naturalmente “espera de I/O”
você usa bibliotecas assíncronas (por exemplo, aiohttp, asyncpg, aiobotocore)

Vantagens

Alto número de tarefas concorrentes com baixo overhead.
Controle explícito de timeouts, cancelamento e backpressure (quando bem projetado).
Evita a complexidade de estado compartilhado entre threads (embora continue havendo concorrência lógica).

Limitações

Se você chamar funções bloqueantes dentro do loop, você “congela” tudo.
CPU-bound dentro do event loop degrada o sistema (a solução é delegar a processos/threads).
Exige disciplina arquitetural: “tudo assíncrono” tende a ser mais simples do que misturar estilos sem planejamento.

Exemplo (I/O concorrente com asyncio + aiohttp)

import asyncio
import aiohttp

URLS = [
    "https://example.com",
    "https://example.org",
    "https://example.net",
]

async def fetch(session: aiohttp.ClientSession, url: str) -> tuple[str, int]:
    async with session.get(url, timeout=aiohttp.ClientTimeout(total=10)) as resp:
        return url, resp.status

async def main():
    async with aiohttp.ClientSession() as session:
        tasks = [fetch(session, url) for url in URLS]
        for coro in asyncio.as_completed(tasks):
            url, status = await coro
            print(url, status)

asyncio.run(main())

6) Como escolher: um “mapa” rápido de decisão

Regra prática por tipo de carga

CPU-bound (cálculo pesado):
multiprocessing (ou ProcessPoolExecutor).
I/O-bound com biblioteca bloqueante (sem versão async):
threading (ou ThreadPoolExecutor).
I/O-bound com stack assíncrona (HTTP async, DB async):
asyncio.

Perguntas que ajudam a decidir

A tarefa passa a maior parte do tempo esperando rede/disco/banco?
Se sim: threads ou asyncio.
A tarefa consome CPU continuamente?
Se sim: processos.
Você precisa de dezenas, centenas ou milhares de tarefas simultâneas?
Se sim: asyncio costuma escalar melhor (desde que tudo seja assíncrono).
Sua biblioteca principal é bloqueante e não há alternativa madura?
Threads podem ser a solução pragmática.
Há necessidade de isolamento de falhas e limites fortes de recursos?
Processos ajudam, mas exigem controle de filas, timeouts e monitoramento.

7) Misturando modelos com segurança (casos reais)

Em sistemas modernos, é comum combinar abordagens:

Asyncio + ProcessPool: servidor async que delega tarefas CPU-bound para processos, mantendo o loop responsivo.
Asyncio + ThreadPool: para chamar um trecho legado bloqueante sem travar o event loop.
Threads + processos: menos comum, mas aparece em pipelines com etapas diferentes.

Exemplo de padrão (asyncio delegando CPU-bound a processos):

import asyncio
from concurrent.futures import ProcessPoolExecutor

def heavy_compute(x: int) -> int:
    total = 0
    for i in range(50_000_00):
        total += (i * x) % 97
    return total

async def main():
    loop = asyncio.get_running_loop()
    with ProcessPoolExecutor() as pool:
        tasks = [loop.run_in_executor(pool, heavy_compute, x) for x in range(1, 5)]
        results = await asyncio.gather(*tasks)
    print(results)

asyncio.run(main())

8) Pontos de Cyber Segurança e confiabilidade (frequentemente esquecidos)

Concorrencia Python não é só performance. Ela afeta diretamente a superfície de falhas e riscos operacionais:

Timeouts e cancelamento: sempre defina timeouts em rede e I/O. Sem isso, você cria negação de serviço “acidental” (recursos presos).
Limites de concorrência: impor rate limiting e semáforos evita saturar:
- conexões de banco
- sockets
- CPU
- limites de API de terceiros
Fila e backpressure: em alto volume, use filas (ex.: asyncio.Queue) e limite de workers.
Logs e rastreabilidade: concorrência aumenta dificuldade de rastrear fluxo. Use IDs de correlação e logs estruturados.
Isolamento: processos isolam melhor falhas e vazamentos de memória; threads compartilham tudo (um bug pode corromper estado global).
Dependências: bibliotecas async imaturas podem ter edge cases; valide maturidade e comportamento sob carga.

Conclusão

Para Concorrencia Python, a escolha correta depende principalmente do tipo de carga:

Threading: melhor para I/O quando você está preso a bibliotecas bloqueantes ou quer uma solução direta.
Multiprocessing: escolha padrão para CPU-bound e paralelismo real em múltiplos núcleos.
Asyncio: excelente para alta escala de I/O, desde que você adote bibliotecas assíncronas e evite chamadas bloqueantes no event loop.

Em termos práticos: identifique se seu gargalo é CPU ou I/O, meça com benchmarks simples, imponha timeouts e limites de concorrência, e selecione o modelo que reduz complexidade sem sacrificar confiabilidade.