Async & Concorrência para Throughput
Modelos de concorrência em Python visam diferentes gargalos: asyncio para paralelismo de I/O, multiprocessing para paralelismo de CPU e threading para I/O com bibliotecas bloqueantes.
Receita
import asyncio
import httpx
async def fetch_all(urls: list[str]) -> list[int]:
async with httpx.AsyncClient() as client:
responses = await asyncio.gather(*[client.get(u) for u in urls])
return [r.status_code for r in responses]Quando usar isso:
- Muitas requisições concorrentes de HTTP/banco de dados (limitado por I/O)
- Trabalho limitado por CPU em múltiplos núcleos (multiprocessing)
- Throughput é mais importante que a latência de requisições individuais
Exemplo de Trabalho
import asyncio
import httpx
from concurrent.futures import ProcessPoolExecutor
# Limitado por I/O: asyncio
async def fetch_urls(urls: list[str]) -> list[str]:
async with httpx.AsyncClient(timeout=10.0) as client:
tasks = [client.get(url) for url in urls]
responses = await asyncio.gather(*tasks, return_exceptions=True)
return [r.text[:100] if isinstance(r, httpx.Response) else str(r) for r in responses]
# Limitado por CPU: pool de processos
def compute(n: int) -> int:
return sum(i * i for i in range(n))
async def run_cpu_bound(items: list[int]) -> list[int]:
loop = asyncio.get_running_loop()
with ProcessPoolExecutor() as pool:
return list(await asyncio.gather(*[
loop.run_in_executor(pool, compute, n) for n in items
]))
async def main():
urls = ["https://httpbin.org/get"] * 20
io_results = await fetch_urls(urls)
cpu_results = await run_cpu_bound([100_000] * 4)
print(len(io_results), sum(cpu_results))
asyncio.run(main())O que isso demonstra:
asyncio.gatherexecuta I/O concorrentemente em uma threadProcessPoolExecutorcontorna o GIL para trabalho de CPUreturn_exceptions=Trueimpede que uma falha cancele todas as outras- Combina modelos: orquestração assíncrona + pool de processos para estágios de CPU
Mergulho Profundo
Seleção de Modelo
| Gargalo | Modelo | Ferramenta |
|---|---|---|
| I/O (rede, disco) | asyncio | httpx, asyncpg |
| CPU | multiprocessing | ProcessPoolExecutor |
| Biblioteca de I/O bloqueante | threading | ThreadPoolExecutor |
| Misto | asyncio + executor | run_in_executor |
Armadilhas
- asyncio para trabalho de CPU - sem ganho de velocidade, adiciona complexidade. Correção: multiprocessing ou vetorização.
- Concorrência ilimitada - sobrecarrega o servidor ou atinge limites de conexão. Correção:
asyncio.Semaphore(50)para limitar requisições paralelas. - Estado mutável compartilhado entre threads - corridas de dados (data races). Correção: filas (queues), locks ou isolamento de processos.
- Criação de processos por tarefa - overhead de fork. Correção: reutilizar ProcessPoolExecutor.
- Chamada bloqueante em função async - congela o loop de eventos. Correção:
await asyncio.to_thread(blocking_fn)ou biblioteca async.
Alternativas
| Alternativa | Usar Quando | Não Usar Quando |
|---|---|---|
| Sync + threading | Código bloqueante legado | Desenvolvimento assíncrono novo (Greenfield async) |
| Celery/RQ | Fila de tarefas distribuída | Paralelismo dentro do processo é suficiente |
| Escalonamento horizontal | Throughput em produção | Fase de otimização local |
FAQs
asyncio ou threads?
asyncio para bibliotecas assíncronas nativas. Threads para bibliotecas bloqueantes que você não pode substituir.
Quantas requisições concorrentes?
Comece com 10-50. Ajuste com semáforo. Monitore os limites do servidor e do cliente.
asyncio contorna o GIL?
Não. asyncio multiplexa I/O em uma thread. Paralelismo de CPU requer multiprocessing.
FastAPI e asyncio?
FastAPI é nativamente assíncrono. Use drivers de banco de dados assíncronos para benefício total.
Como limito a taxa?
asyncio.Semaphore ou token bucket com asyncio.sleep.
Tamanho do pool de processos?
Geralmente os.cpu_count(). Pools de I/O podem ser maiores que a contagem de CPUs.
asyncio.gather vs TaskGroup?
Python 3.14: prefira asyncio.TaskGroup para concorrência estruturada e tratamento de erros.
Como faço benchmark de throughput?
Requisições por segundo sob carga (locust, hey, k6). Não tempo de requisição única com timeit.
Segurança de thread com free-threaded 3.14?
Audite estado mutável compartilhado. Use locks ou estruturas de dados imutáveis.
Quando sync é suficiente?
Baixo tráfego, poucas operações de I/O concorrentes, ou jobs em lote sem requisitos de latência.
Relacionado
- 30 Regras Async - armadilhas do asyncio
- Concorrência - GIL e modelos
- Cache e Memoização - reduzir I/O
- Melhores Práticas de Performance - seleção de modelo
Versões da Stack: Esta página foi escrita para Python 3.14.0, FastAPI 0.115+, Django 5.2, Flask 3.1, Pydantic 2, PyTorch 2.6+, pandas 2.2+, Polars 1.x, ruff 0.9+, e uv 0.6+.