asyncio ou threading? Escolha certo em 3 perguntas

1. Visão geral

Muitas pessoas que estão aprendendo Python concurrency perguntam “asyncio ou threading é melhor?” — essa pergunta está errada desde o início, porque as duas ferramentas resolvem problemas diferentes. Este artigo apresenta 3 perguntas específicas para você mesmo responder, acompanhadas de código funcionável. Depois de ler, você saberá exatamente quando usar threading, quando usar asyncio, e quando ambos são inúteis (deve usar multiprocessing).

2. Por que a pergunta “qual é mais rápido” está errada

Python tem Global Interpreter Lock (GIL) — apenas uma thread Python executa bytecode de cada vez, não importa quantos núcleos de CPU a máquina tem. threading e asyncio não conseguem quebrar o GIL. Ambos apenas ajudam a CPU a não “travar” quando espera por I/O (leitura de arquivo, chamada de rede, consulta de DB) — eles não ajudam cálculos a ficarem mais rápidos.

import time

def cpu_bound_task():
    total = 0
    for i in range(10_000_000):
        total += i * i
    return total

start = time.perf_counter()
cpu_bound_task()
cpu_bound_task()
print(f"Sequential: {time.perf_counter() - start:.2f}s")

Sequential: 1.85s

Mesmo que você execute duas vezes cpu_bound_task() usando threading.Thread ou asyncio.gather, o tempo total não diminui, porque o GIL só permite que uma thread execute código Python de cada vez. Essa é a razão pela qual a primeira pergunta abaixo é a mais importante.

Nota: Se o problema é CPU-bound (cálculo matemático pesado, processamento de imagem, criptografia…), nem threading nem asyncio ajudam. Você precisa de multiprocessing ou escrever a parte pesada em uma extensão C/Rust (NumPy, Cython).

3. Pergunta 1: Seu trabalho é I/O-bound ou CPU-bound?

Esta é a pergunta que determina se você deve usar concorrência do tipo “aguardar” (I/O-bound) ou não.

3.1. I/O-bound: tanto threading quanto asyncio são eficientes

import time
import requests
from concurrent.futures import ThreadPoolExecutor

URLS = ["https://httpbin.org/delay/1"] * 5

def fetch(url: str) -> int:
    return requests.get(url).status_code

start = time.perf_counter()
with ThreadPoolExecutor(max_workers=5) as pool:
    results = list(pool.map(fetch, URLS))
print(f"Threading: {time.perf_counter() - start:.2f}s, results={results}")

Threading: 1.12s, results=[200, 200, 200, 200, 200]

ThreadPoolExecutor executa 5 requisições simultaneamente usando 5 threads do SO. Como requests.get é uma chamada bloqueante esperando por rede, o GIL é liberado quando a thread está esperando I/O na camada C — então 5 requisições rodam em paralelo de verdade, tempo total ~1s em vez de 5s.

import asyncio
import aiohttp
import time

URLS = ["https://httpbin.org/delay/1"] * 5

async def fetch(session: aiohttp.ClientSession, url: str) -> int:
    async with session.get(url) as resp:
        return resp.status

async def main():
    async with aiohttp.ClientSession() as session:
        tasks = [fetch(session, url) for url in URLS]
        return await asyncio.gather(*tasks)

start = time.perf_counter()
results = asyncio.run(main())
print(f"Asyncio: {time.perf_counter() - start:.2f}s, results={results}")

Asyncio: 1.08s, results=[200, 200, 200, 200, 200]

Resultado equivalente ao threading, mas o mecanismo é diferente: apenas uma thread única, o event loop controla ativamente await e cede o controle quando espera I/O, não precisa que o SO crie novas threads.

3.2. CPU-bound: ambas perdem para multiprocessing

from concurrent.futures import ProcessPoolExecutor

def cpu_bound_task():
    return sum(i * i for i in range(10_000_000))

start = time.perf_counter()
with ProcessPoolExecutor(max_workers=2) as pool:
    results = list(pool.map(cpu_bound_task, [None, None]))
print(f"Multiprocessing: {time.perf_counter() - start:.2f}s")

Multiprocessing: 1.02s

ProcessPoolExecutor cria processes separados, cada process tem seu próprio GIL, então consegue aproveitar vários núcleos de CPU de verdade. Se você está em dúvida entre “threading ou asyncio” para um problema CPU-bound, a resposta correta é ambos estão errados — use multiprocessing.

4. Pergunta 2: O ecossistema que você usa oferece suporte a async?

Esta é a pergunta mais prática quando você trabalha em codebase real, onde você não escreve tudo do zero.

import psycopg2  # driver síncrono, bloqueante

def get_user(user_id: int) -> dict:
    conn = psycopg2.connect("dbname=app")
    cur = conn.cursor()
    cur.execute("SELECT id, name FROM users WHERE id = %s", (user_id,))
    row = cur.fetchone()
    conn.close()
    return {"id": row[0], "name": row[1]}

psycopg2 é uma biblioteca bloqueante — não tem versão await. Para rodar várias queries em paralelo com este driver, threading (ou run_in_executor do asyncio) é a escolha natural, porque você não precisa reescrever nada.

import asyncpg  # driver nativo async

async def get_user(conn: asyncpg.Connection, user_id: int) -> dict:
    row = await conn.fetchrow(
        "SELECT id, name FROM users WHERE id = $1", user_id
    )
    return {"id": row["id"], "name": row["name"]}

Se seu ecossistema já tem driver nativo async (asyncpg, aiohttp, motor para MongoDB…), asyncio aproveita ao máximo sua vantagem: milhares de conexões simultâneas em uma única thread, sem o custo de criar threads do SO.

Nota: Misturar código bloqueante dentro de async def sem await corretamente (por exemplo, chamar requests.get diretamente dentro de uma coroutine) vai bloquear todo o event loop — todas as outras tasks têm que esperar, até mesmo tasks não relacionadas. Este é o erro mais comum ao migrar de threading para asyncio.

5. Pergunta 3: Quantas tasks simultâneas você precisa?

O custo de recursos entre uma thread e uma coroutine diferencia muito quando o número de tasks aumenta.

import threading
import resource

def noop():
    threading.Event().wait(2)

threads = [threading.Thread(target=noop) for _ in range(2000)]
for t in threads:
    t.start()

peak_kb = resource.getrusage(resource.RUSAGE_SELF).ru_maxrss
print(f"Peak RSS com 2000 threads: {peak_kb / 1024:.1f} MB")

for t in threads:
    t.join()

Peak RSS com 2000 threads: 1850.3 MB

Cada thread do SO por padrão ocupa alguns centenas de KB até alguns MB de stack, então 2000 threads facilmente consomem perto de 2 GB de RAM e colocam pressão de context-switching grande no scheduler do SO.

import asyncio
import resource

async def noop():
    await asyncio.sleep(2)

async def main():
    tasks = [asyncio.create_task(noop()) for _ in range(2000)]
    await asyncio.gather(*tasks)
    peak_kb = resource.getrusage(resource.RUSAGE_SELF).ru_maxrss
    print(f"Peak RSS com 2000 coroutines: {peak_kb / 1024:.1f} MB")

asyncio.run(main())

Peak RSS com 2000 coroutines: 95.7 MB

Coroutine é um objeto Python puro, não precisa de thread separada do SO — o custo de criar 2000 tasks é apenas dezenas de MB. Se você precisa lidar com milhares de conexões simultâneas (web server, chat server, crawler em larga escala), asyncio é praticamente uma escolha obrigatória; threading vai ficar sem recursos antes de ficar sem largura de banda.

6. Tabela de decisão rápida

Pergunta	Resposta → Escolha
1. I/O-bound ou CPU-bound?	CPU-bound → multiprocessing. I/O-bound → passe para pergunta 2
2. A biblioteca que você usa tem async-native?	Não (driver bloqueante antigo) → threading. Sim (aiohttp, asyncpg…) → asyncio
3. Quantas tasks simultâneas você precisa?	Algumas dezenas → threading é simples o suficiente. Centenas/milhares → asyncio

Na prática, muitos sistemas usam ambos: asyncio para a camada de network I/O-bound em larga escala, combinado com run_in_executor para rodar pedaços de código bloqueante (por exemplo biblioteca antiga) em um thread pool separado para não bloquear o event loop.

import asyncio

def legacy_blocking_call(x: int) -> int:
    import time
    time.sleep(1)  # simula biblioteca bloqueante antiga
    return x * 2

async def main():
    loop = asyncio.get_running_loop()
    result = await loop.run_in_executor(None, legacy_blocking_call, 21)
    print(result)

asyncio.run(main())

42

run_in_executor move a função bloqueante para uma thread no thread pool padrão, o event loop continua livre para lidar com outras coroutines enquanto aguarda.

7. Testando comportamento com pytest

Exemplo de teste confirmando que asyncio.gather executa as tasks em paralelo de verdade, não sequencialmente:

import asyncio
import time
import pytest

async def sleepy_task(seconds: float) -> float:
    await asyncio.sleep(seconds)
    return seconds

@pytest.mark.asyncio
async def test_gather_runs_concurrently():
    start = time.perf_counter()
    results = await asyncio.gather(
        sleepy_task(0.5), sleepy_task(0.5), sleepy_task(0.5)
    )
    elapsed = time.perf_counter() - start

    assert results == [0.5, 0.5, 0.5]
    # Se rodar sequencialmente levaria ~1.5s, rodando em paralelo apenas ~0.5s
    assert elapsed < 0.7

Dependências necessárias para o teste:

pytest
pytest-asyncio

Este teste confirma a verdadeira natureza de asyncio: três tasks sleep(0.5) rodando simultâneas em uma única thread levam apenas ~0.5s no total, não 1.5s — se a assertion elapsed < 0.7 falhar, significa que seu código está fazendo await sequencialmente em vez de usar gather.

8. Erros comuns

• Usar time.sleep() dentro de uma coroutine: deve usar await asyncio.sleep(). time.sleep() é uma chamada bloqueante, congela todo o event loop.
• Esquecer await: chamar fetch(session, url) sem await apenas cria um objeto coroutine, não executa nada — Python vai avisá-lo com RuntimeWarning: coroutine was never awaited.
• Compartilhar estado entre threads sem lock: threading tem race condition real (múltiplas threads escrevendo a mesma variável), precisa de threading.Lock. asyncio em uma única thread não tem race condition deste tipo, mas pode ter bugs de lógica se await estiver no meio do que parecia ser código “atomic”.
• Usar ThreadPoolExecutor para problema CPU-bound e se decepcionar porque não fica mais rápido — porque o GIL ainda bloqueia, use ProcessPoolExecutor.

Nota: Se seu time já está acostumado com threading e o codebase é todo biblioteca bloqueante, não se apresse em refatorar para asyncio só porque é “mais novo”. O custo de reescrever todos os drivers, lidar com exception diferentemente, e debugar stack trace de coroutine bem mais complexo é bem maior que o benefício obtido se o número de tasks simultâneas não for grande.

9. Conclusão

Escolher entre asyncio e threading não é sobre “qual é mais rápido” mas sobre três perguntas específicas: o problema é I/O-bound ou CPU-bound, o ecossistema de biblioteca tem async-native ou não, e qual escala de concorrência você precisa lidar. CPU-bound sempre precisa de multiprocessing. I/O-bound com biblioteca antiga e número pequeno de tasks, então threading é simples e fácil de integrar. I/O-bound em larga escala com biblioteca async-native, então asyncio é a escolha correta em termos de performance e recursos.