Guia completo do Python asyncio: dominando a simultaneidade em 2026

O que o asyncio realmente resolve

asyncio existe para um trabalho: executar muitas operações vinculadas a E/S simultaneamente em um único thread, sem a sobrecarga de threads ou processos. Quando seu programa passa a maior parte do tempo esperando — por uma resposta do banco de dados, uma resposta HTTP, uma leitura de arquivo — o asyncio permite que ele alterne para outro trabalho durante essa espera, em vez de ficar ocioso.

Isso é fundamentalmente diferente do paralelismo vinculado à CPU. asyncio não oferece mais núcleos de CPU; oferece melhor utilização de um único núcleo enquanto aguarda E/S. Para trabalhos pesados ​​de CPU (processamento de imagens, processamento de números), você ainda desejamultiprocessingou uma extensão nativa.

O loop de eventos, explicado

No centro de cada programa assíncrono está o loop de eventos – um agendador de thread único que executa corrotinas, descontrola quando se espera algo e o retoma quando o resultado esperado está pronto.

import asyncio

async def say_hello():
    print("start")
    await asyncio.sleep(1)
    print("end")

asyncio.run(say_hello())

asyncio.run()cria um novo loop de eventos, executa a corrotina até a conclusão e fecha o loop. É o ponto de entrada de nível superior correto para qualquer script assíncrono – evite o antigoget_event_loop().run_until_complete()padrão, a menos que você tenha um motivo específico para gerenciar o loop manualmente.

Corrotinas vs Tarefas

Um objeto de corrotina (criado chamando uma função ||||) não faz nada por si só – é um objeto semelhante a um gerador pausado esperando para ser acionado. Aguardando, ele é executado em linha, sequencialmente. Umaasync defTarefaenvolve uma corrotina e a agenda para ser executada simultaneamente no loop de eventos imediatamente.tarefas.py

import asyncio
import time

async def fetch(n):
    await asyncio.sleep(1)
    return n * 2

async def sequential():
    start = time.perf_counter()
    results = [await fetch(i) for i in range(5)]
    print("sequential:", time.perf_counter() - start)
    return results

async def concurrent():
    start = time.perf_counter()
    tasks = [asyncio.create_task(fetch(i)) for i in range(5)]
    results = await asyncio.gather(*tasks)
    print("concurrent:", time.perf_counter() - start)
    return results

asyncio.run(sequential())   # ~5 seconds
asyncio.run(concurrent())   # ~1 second

reunir() vs Grupo de Tarefas

tem sido a maneira padrão de executar várias corrotinas simultaneamente desde o Python 3.4, mas tem uma vantagem: se uma tarefa for levantada, as outras continuarão sendo executadas em segundo plano, a menos que você passe

asyncio.gather()e lidar com erros manualmente. Essa é uma fonte comum de tarefas órfãs e avisos de exceção não tratadas nos logs de produção.return_exceptions=TruePython 3.11 introduzido

, uma primitiva de simultaneidade estruturada que corrige isso corretamente: se alguma tarefa filha falhar, o grupo cancela todas as tarefas irmãs restantes e gera umasyncio.TaskGroupcontendo todas as falhas.ExceptionGroupgrupo de tarefas.py

import asyncio

async def risky(n):
    if n == 2:
        raise ValueError("boom at " + str(n))
    await asyncio.sleep(1)
    return n

async def main():
    try:
        async with asyncio.TaskGroup() as tg:
            results = [tg.create_task(risky(i)) for i in range(4)]
    except* ValueError as eg:
        for exc in eg.exceptions:
            print("caught:", exc)

asyncio.run(main())

identificadores de sintaxeexcept*instâncias, permitindo processar todas as falhas que ocorreram no grupo, em vez de apenas a primeira. Se você deseja Python 3.11 ou posterior, prefiraExceptionGroupmais cruTaskGrouppara qualquer coisa além da simultaneidade trivial de disparar e esquecer.gather()Gerenciadores e bloqueios de contexto assíncrono

O estado mutável compartilhado entre tarefas simultâneas ainda precisa de proteção, mesmo em um único thread, porque um

point pode entregar o controle para outra tarefa no meio da operação.awaitprotege seções críticas da mesma maneiraasyncio.Lockfaz para threads.threading.Locklock_example.py

import asyncio

counter = 0
lock = asyncio.Lock()

async def increment():
    global counter
    async with lock:
        current = counter
        await asyncio.sleep(0)  # simulate a yield point
        counter = current + 1

async def main():
    await asyncio.gather(*(increment() for _ in range(100)))
    print(counter)  # always 100, never less

asyncio.run(main())

o ponto de rendimento permitiria que outra tarefa se intercalasse entre a leitura e a escritaawait asyncio.sleep(0), causando perda de atualizações — a condição de corrida clássica, apenas em um thread em vez de muitos.counterA armadilha de bloqueio de chamadas

Esta é a maior fonte de relatórios de bugs “meu aplicativo assíncio é lento”. O loop de eventos é de thread único. Qualquer chamada síncrona que bloqueie —

, um bloqueiotime.sleep(), um loop com uso intenso de CPU, um arquivo sem buffer lido em um disco lento — congela o loop inteiro, incluindo todas as outras tarefas que aguardam nele.requests.get()ruim_vs_bom.py

# BAD: blocks the whole event loop for 2 seconds
import time
async def bad():
    time.sleep(2)

# GOOD: yields control back to the loop
import asyncio
async def good():
    await asyncio.sleep(2)

# GOOD: offload a real blocking call to a thread
async def good_blocking_lib():
    result = await asyncio.to_thread(some_blocking_function, arg1, arg2)
    return result

(3.9+) para bloquear E/S de bibliotecas de terceiros que não possuem um equivalente assíncrono. Para trabalho vinculado à CPU, useasyncio.to_thread()com umloop.run_in_executor()em vez disso – os threads não ajudarão no trabalho vinculado à CPU por causa do GIL.ProcessPoolExecutorTempos limite e cancelamento

O código de produção precisa de tempos limite em todas as chamadas externas. Python 3.11 adicionado

Production code needs timeouts on every external call. Python 3.11 addedasyncio.timeout()como um substituto limpo do gerenciador de contexto para o antigoasyncio.wait_for().

import asyncio

async def slow_call():
    await asyncio.sleep(10)

async def main():
    try:
        async with asyncio.timeout(2):
            await slow_call()
    except TimeoutError:
        print("operation timed out after 2s")

asyncio.run(main())

Quando um tempo limite é acionado, o asyncio cancela a tarefa interna aumentandoasyncio.CancelledErrordentro dele. Se sua corrotina agrupa recursos que precisam de limpeza (conexões abertas, identificadores de arquivos), usetry/finallyou um gerenciador de contexto assíncrono para que o cancelamento não vaze recursos.

Exemplo real: buscador de API simultâneo

Aqui está um exemplo completo em formato de produção: busca de vários URLs simultaneamente com um limite de conexão, um tempo limite por solicitação e tratamento de erros estruturado usandoaiohttp.

import asyncio
import aiohttp

async def fetch_one(session, url, semaphore):
    async with semaphore:
        try:
            async with asyncio.timeout(5):
                async with session.get(url) as resp:
                    return url, resp.status, await resp.text()
        except (TimeoutError, aiohttp.ClientError) as e:
            return url, None, str(e)

async def fetch_all(urls, max_concurrent=10):
    semaphore = asyncio.Semaphore(max_concurrent)
    async with aiohttp.ClientSession() as session:
        async with asyncio.TaskGroup() as tg:
            tasks = [tg.create_task(fetch_one(session, u, semaphore)) for u in urls]
        return [t.result() for t in tasks]

urls = ["https://example.com"] * 50
results = asyncio.run(fetch_all(urls))
print(len(results), "requests completed")

OSemaphorelimita quantas solicitações são executadas ao mesmo tempo, o que protege tanto o cliente quanto o servidor de destino de serem sobrecarregados por centenas de conexões simultâneas. Esse padrão — reutilização de sessão, simultaneidade limitada por semáforo, tempo limite por solicitação, grupo de tarefas estruturado — é próximo ao que você veria em um raspador de produção ou agregador de API.

Erros Comuns

• Esquecendo de aguardar uma corrotina.Chamando umasync deffunção semawaitapenas cria um objeto de corrotina e não faz nada – o Python avisará “a corrotina nunca foi aguardada”, mas é fácil perdê-la em logs barulhentos.
• Criar tarefas sem manter uma referência. asyncio.create_task()contém apenas uma referência fraca internamente; se o objeto de tarefa for coletado como lixo antes de terminar, ele poderá ser cancelado silenciosamente. Armazene tarefas em uma lista ou conjunto até que sejam concluídas.
• Misturando drivers de banco de dados sincronizados e assíncronos.Usar um driver de bloqueio (como simplespsycopg2) dentro do código assíncrono anula o propósito – use um driver assíncrono (asyncpg,aiomysql) ou envolva a chamada comto_thread().
• Não definir tempos limite.Uma única chamada de rede interrompida sem tempo limite pode paralisar indefinidamente um manipulador de solicitações inteiro sob carga.
• Uso excessivo de assíncio para trabalho vinculado à CPU.Se o seu gargalo é a computação, e não a espera, o asyncio adiciona complexidade sem adicionar velocidade.