Asyncio em Python: async/await, TaskGroup, threads, multiprocessing e GIL

Asyncio é a biblioteca embutida do Python para código assíncrono. As funções declaradas com async def sabem se suspender no await, e enquanto uma delas espera a resposta da rede, o laço de eventos executa outras. Assim, centenas de requisições seguem ao mesmo tempo em um único thread. Mas o asyncio é apenas uma das três ferramentas: há também threads (threading) e processos (multiprocessing). Vamos ver como o asyncio em Python funciona, o que escolher para cada tarefa e qual o papel do GIL.

Esperar e calcular são tarefas diferentes

Sem essa distinção nada do resto fica claro. Um programa fica lento por um de dois motivos:

• I/O-bound — espera entrada e saída: a resposta de uma API, de um banco de dados, a leitura de disco. O processador fica ocioso.
• CPU-bound — faz cálculos: processa imagens, percorre laços com milhões de registros. O processador está carregado.

Para esperar servem o asyncio e as threads: enquanto uma operação aguarda, outra é executada. Para cálculos são necessárias computações paralelas de verdade em vários núcleos — processos. Confundir os dois é o erro principal: threads não aceleram cálculos em Python puro, e processos para três requisições HTTP são um canhão para matar uma mosca.

O que é asyncio: corrotinas e laço de eventos

O código assíncrono se apoia em duas palavras. async def declara uma corrotina — uma função capaz de se suspender. Chamá-la não executa o corpo, apenas cria um objeto de corrotina. O await dentro dela significa: «aqui eu espero, e você, enquanto isso, cuida de outras tarefas». Tudo é controlado pelo laço de eventos (event loop), iniciado com asyncio.run().

import asyncio

async def fetch_rate(currency):
    await asyncio.sleep(1)          # simulação de uma requisição à API
    return {"USD": 92.5, "EUR": 100.1, "CNY": 12.7}[currency]

async def main():
    rate = await fetch_rate("USD")
    print(rate)

asyncio.run(main())  # 92.5

asyncio.sleep() é uma pausa assíncrona: ela devolve o controle ao laço de eventos. Já o time.sleep() comum dentro de uma corrotina bloqueia todo o laço, e as demais tarefas ficam paradas.

Em sequência ou ao mesmo tempo

Se você escrever await dentro de um laço, as requisições irão uma após a outra — três moedas levarão três segundos. Para que sejam feitas simultaneamente, é preciso iniciá-las como tarefas.

import asyncio, time

async def main():
    start = time.perf_counter()
    for cur in ("USD", "EUR", "CNY"):
        await fetch_rate(cur)                       # uma por vez
    print(f"{time.perf_counter() - start:.1f} s")   # 3.0 s

asyncio.run(main())

TaskGroup: iniciar juntas e esperar todas

A forma moderna (Python 3.11+) é o grupo de tarefas. Dentro do bloco async with cada tarefa começa imediatamente por meio de tg.create_task(), e a saída do bloco acontece quando todas terminam.

async def main():
    start = time.perf_counter()
    async with asyncio.TaskGroup() as tg:
        tasks = {cur: tg.create_task(fetch_rate(cur)) for cur in ("USD", "EUR", "CNY")}
    rates = {cur: t.result() for cur, t in tasks.items()}
    print(rates, f"{time.perf_counter() - start:.1f} s")   # ... 1.0 s

asyncio.run(main())

O TaskGroup age de forma rigorosa: se uma tarefa falha, as outras são canceladas, e todos os erros são reunidos em um ExceptionGroup. Capturam-se eles com except*:

async def main():
    try:
        async with asyncio.TaskGroup() as tg:
            tg.create_task(fetch_rate("USD"))
            tg.create_task(fetch_rate("XXX"))   # KeyError
    except* KeyError as eg:
        print("moeda desconhecida:", eg.exceptions)

asyncio.gather e create_task

Antes do TaskGroup o padrão era asyncio.gather(): ele executa as corrotinas simultaneamente e devolve os resultados na mesma ordem. Ele ainda aparece com frequência em código existente.

async def main():
    usd, eur, cny = await asyncio.gather(
        fetch_rate("USD"), fetch_rate("EUR"), fetch_rate("CNY")
    )
    # com return_exceptions=True os erros voltam na lista, sem interromper tudo

asyncio.create_task() inicia uma corrotina em segundo plano, sem grupo. A referência à tarefa precisa ser guardada, senão o coletor de lixo pode removê-la, e ninguém verá os erros. Por isso, em projetos novos o TaskGroup é mais confiável.

Cancelamento de tarefas e timeouts

async def main():
    task = asyncio.create_task(fetch_rate("USD"))
    task.cancel()                     # dentro da tarefa surgirá CancelledError
    try:
        await task
    except asyncio.CancelledError:
        print("cancelada")

    try:
        async with asyncio.timeout(0.5):   # Python 3.11+
            await fetch_rate("EUR")
    except TimeoutError:
        print("demorou demais")

Em versões antigas, no lugar de asyncio.timeout() usa-se asyncio.wait_for(coro, timeout=0.5). Existe ainda o asyncio.wait(), que devolve conjuntos de tarefas concluídas e pendentes e sabe esperar a primeira (return_when=asyncio.FIRST_COMPLETED).

No Python 3.15, como mostra o curso, o TaskGroup ganhou o método tg.cancel(): ele cancela todas as tarefas do grupo de uma vez. Isso resolve o cenário «quem responder primeiro»: consultar três serviços espelho, pegar a primeira resposta e cancelar as demais. Antes era preciso lançar uma exceção-sinal própria ou cancelar as tarefas manualmente.

Sincronização: Semaphore, Lock, Event, Queue

O asyncio tem primitivas próprias que funcionam com await. A mais útil é o Semaphore: ele limita a quantidade de operações simultâneas, para não derrubar uma API alheia com mil requisições.

sem = asyncio.Semaphore(10)

async def limited(url):
    async with sem:              # no máximo 10 ao mesmo tempo
        await asyncio.sleep(0.1) # aqui estaria a requisição
        return url

• asyncio.Lock — apenas uma corrotina na seção crítica.
• asyncio.Event — uma corrotina espera em await event.wait(), e outra sinaliza com event.set().
• asyncio.Queue — fila «produtor — consumidor»: await q.put(), await q.get(), q.task_done().

Armadilha: esqueceram o await

O erro mais comum é chamar uma corrotina sem await:

async def main():
    rate = fetch_rate("USD")    # esqueceram o await
    print(rate * 2)             # TypeError
# RuntimeWarning: coroutine 'fetch_rate' was never awaited

O Python não executou a função, apenas criou o objeto da corrotina. Se nos logs aparece «was never awaited», procure o await esquecido. O segundo erro típico é uma chamada bloqueante (time.sleep, requests.get) dentro de uma corrotina. Para HTTP use bibliotecas assíncronas (httpx, aiohttp), e o código síncrono inevitável transfira para uma thread com await asyncio.to_thread(func, arg).

Multithreading em Python: o módulo threading

A thread é uma linha de execução independente dentro de um mesmo processo, e as threads compartilham a memória. Para esperar a rede funcionam muito bem e não exigem reescrever o código em async.

import threading, time

def download(name):
    time.sleep(1)               # espera pela rede
    print(name, "pronto")

threads = [threading.Thread(target=download, args=(n,)) for n in ("a", "b", "c")]
for t in threads:
    t.start()
for t in threads:
    t.join()                    # tudo em cerca de 1 segundo

A memória compartilhada é ao mesmo tempo vantagem e fonte de erros: se duas threads alteram a mesma variável, é preciso um threading.Lock. Em geral é mais simples não lidar com threads manualmente e usar um pool.

ThreadPoolExecutor

concurrent.futures.ThreadPoolExecutor cria um pool de threads e distribui as tarefas entre elas. O map devolve os resultados na ordem original.

from concurrent.futures import ThreadPoolExecutor
import time

def fetch(url):
    time.sleep(1)
    return f"{url}: 200"

urls = [f"https://example.com/{i}" for i in range(5)]
with ThreadPoolExecutor(max_workers=5) as pool:
    for result in pool.map(fetch, urls):
        print(result)           # tudo em cerca de 1 segundo

O que é o GIL e por que threads não aceleram cálculos

GIL (Global Interpreter Lock) é a trava global do interpretador CPython. A cada momento, o bytecode do Python é executado por apenas uma thread. Enquanto uma thread espera entrada e saída, ela libera o GIL, por isso as threads são úteis para a rede. Já em cálculos puros, quatro threads não são mais rápidas que uma: elas se revezam passando a trava umas às outras.

from concurrent.futures import ThreadPoolExecutor

def heavy(n):
    return sum(i * i for i in range(n))

with ThreadPoolExecutor(4) as pool:
    list(pool.map(heavy, [5_000_000] * 4))   # praticamente o mesmo que em sequência

Bibliotecas escritas em C (NumPy e similares) muitas vezes liberam o GIL durante operações pesadas, por isso, para elas, as threads trazem ganho de velocidade.

multiprocessing: paralelismo de verdade

Cada processo tem seu próprio interpretador e seu próprio GIL, por isso os processos calculam em núcleos diferentes ao mesmo tempo. O preço é a memória separada: os dados entre processos são transmitidos por serialização (pickle), e iniciar um processo custa mais que uma thread.

from concurrent.futures import ProcessPoolExecutor

def heavy(n):
    return sum(i * i for i in range(n))

if __name__ == "__main__":      # obrigatório no Windows e no macOS
    with ProcessPoolExecutor() as pool:
        print(list(pool.map(heavy, [5_000_000] * 4)))

Sem a verificação if __name__ == "__main__", os processos filhos, ao importar o módulo, começariam a criar novos processos. Existe também o módulo de baixo nível multiprocessing, com Process, Pool, Queue, mas o ProcessPoolExecutor é mais simples e tem a mesma interface do pool de threads.

O que escolher: asyncio, threading ou multiprocessing

Python sem GIL e subinterpretadores: novidades

A partir do Python 3.13 existe uma build sem GIL (free-threading, PEP 703) — ela é identificada pela letra t no nome, por exemplo python3.15t. Na 3.13 era experimental; a partir da 3.14 (PEP 779) tem suporte oficial. Nela as threads calculam em paralelo em núcleos diferentes. Mas por padrão você continua obtendo o Python comum, com GIL: código de uma thread na build free-threaded roda um pouco mais devagar, e nem todas as bibliotecas estão preparadas para ela.

import sys
print(sys._is_gil_enabled())   # False na build free-threaded

Um segundo caminho para cálculos paralelos surgiu na 3.14 (PEP 734) — os subinterpretadores, no módulo concurrent.interpreters: vários interpretadores isolados dentro de um processo, cada um com seu GIL. É mais cômodo usá-los pelo InterpreterPoolExecutor, que tem a mesma aparência de um pool de threads. Na medição do curso, quatro tarefas pesadas levaram 0,39 s contra 1,03 s em uma thread só.

Resumo

O asyncio em Python é uma ferramenta para tarefas de espera: corrotinas com async def, pausas com await e execução simultânea por meio do TaskGroup ou do gather. As threads resolvem o mesmo problema sem reescrever o código em async, e para cálculos pesados são necessários processos, porque o GIL não permite que as threads do CPython comum calculem em paralelo. O free-threading e os subinterpretadores mudam aos poucos esse quadro, mas a regra básica continua valendo: «esperar — asyncio ou threads; calcular — processos». No curso tudo isso é praticado: o controle de gastos obtém as cotações das moedas em paralelo, em um segundo em vez de três.