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

URL: https://skillaria.top/pt/guides/python/asyncio-python-async-await

Asyncio é a biblioteca embutida do Python para código assíncrono: funções declaradas com async def (corrotinas) são suspensas no await, e enquanto uma espera a resposta da rede, o laço de eventos executa outras tarefas, por isso centenas de requisições podem ocorrer ao mesmo tempo em um só thread. Para aguardar rede e disco servem asyncio ou threads (threading); para cálculos pesados — processos (multiprocessing), porque o GIL não permite que threads do CPython comum calculem em paralelo.

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

| Tarefa | Ferramenta | Por quê |
| --- | --- | --- |

| Baixar 200 páginas, muitas requisições a APIs | asyncio (httpx/aiohttp) ou ThreadPoolExecutor | espera pela rede, processador livre |

| Várias requisições dentro de um programa já assíncrono | TaskGroup | o código já está em async |

| Biblioteca síncrona sem versão async | threads, `asyncio.to_thread` | não é preciso reescrever o código |

| Cálculo pesado em Python puro | ProcessPoolExecutor | contorna o GIL, usa todos os núcleos |

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

## FAQ

**Dá para chamar uma função async a partir de código comum?**

Sim, com asyncio.run(coro()) — ele cria o laço de eventos, executa a corrotina e fecha o laço. Dentro de um laço já em execução, asyncio.run não pode ser chamado; ali se usa await.

**O asyncio funciona com vários threads?**

Não, o laço de eventos executa todas as corrotinas em um único thread e alterna entre elas apenas nos pontos de await. Por isso há menos condições de corrida do que com threads, mas uma operação bloqueante para tudo.

**Preciso instalar o asyncio pelo pip?**

Não, o asyncio faz parte da biblioteca padrão do Python 3; basta escrever import asyncio. O pacote asyncio no PyPI está desatualizado e não deve ser instalado.

**Como executar uma função pesada dentro do asyncio sem travar o laço?**

Use loop.run_in_executor com ProcessPoolExecutor para cálculos, ou asyncio.to_thread para entrada e saída bloqueante; o laço de eventos continua atendendo as demais tarefas.

**Em que a corrotina difere de uma tarefa (Task)?**

A corrotina é um objeto que não faz nada até ser aguardada com await. A tarefa envolve a corrotina e a agenda imediatamente no laço de eventos, por isso ela roda em paralelo com o restante do código.


---
Source: Skillaria.Top (https://skillaria.top)