Bem-vindo à aula 64 do nosso curso de Python! Hoje vamos mergulhar em um dos tópicos mais poderosos e, ao mesmo tempo, mais desafiadores da linguagem: a programação assíncrona com o módulo asyncio. Se você já se deparou com código que trava por causa de operações de I/O, como leitura de arquivos ou requisições web, entender async e await será um divisor de águas na sua carreira.

Nesta aula, você vai aprender os fundamentos do asyncio, como escrever funções assíncronas com async def e como pausar e retomar a execução com await. Também exploraremos o coração do asyncio: o event loop. Ao final, você saberá identificar situações em que a programação assíncrona é a melhor escolha e quando ela pode ser desnecessária ou até prejudicial.

asyncio

O módulo asyncio, introduzido na versão 3.4 do Python, é a biblioteca padrão para escrever código concorrente usando a sintaxe async/await. Ele fornece uma infraestrutura para executar tarefas de forma assíncrona, ou seja, sem bloquear o thread principal enquanto aguarda operações que demoram, como chamadas de rede, leitura de arquivos ou temporizadores.

A ideia central do asyncio é permitir que um programa realize várias operações de I/O simultaneamente, aproveitando o tempo de espera para executar outras tarefas. Isso é diferente de programação paralela (com threads ou processos), pois aqui tudo roda em um único thread, mas de forma cooperativa: cada tarefa cede o controle quando está ociosa, permitindo que outras progridam.

Para começar, vamos ver um exemplo simples de uma função que simula uma operação demorada usando asyncio.sleep():

import asyncio

async def minha_tarefa():
    print("Início")
    await asyncio.sleep(2)  # Simula uma operação que leva 2 segundos
    print("Fim")

asyncio.run(minha_tarefa())

Note que usamos async def para definir a função e await para esperar a conclusão do sleep. A função asyncio.run() é o ponto de entrada que executa a corrotina e gerencia o event loop para nós. Este é o padrão mais básico de uso do asyncio.

async def e await

Em Python, qualquer função definida com async def é chamada de corrotina. Quando você chama uma corrotina, ela não executa imediatamente; em vez disso, retorna um objeto de corrotina que precisa ser aguardado ou agendado no event loop. A palavra-chave await é usada dentro de uma corrotina para pausar sua execução até que outra corrotina ou operação assíncrona seja concluída.

Vamos entender melhor com um exemplo. Suponha que você tenha duas funções que simulam downloads de arquivos:

import asyncio

async def download_arquivo(nome, tempo):
    print(f"Iniciando download de {nome}...")
    await asyncio.sleep(tempo)
    print(f"Download de {nome} concluído!")
    return f"{nome} baixado"

async def main():
    # Executa as duas tarefas concorrentemente
    tarefa1 = asyncio.create_task(download_arquivo("arquivo1.txt", 3))
    tarefa2 = asyncio.create_task(download_arquivo("arquivo2.txt", 1))

    resultado1 = await tarefa1
    resultado2 = await tarefa2

    print(resultado1)
    print(resultado2)

asyncio.run(main())

Neste código, asyncio.create_task() agenda a execução das corrotinas no event loop, e await espera cada uma terminar. Como as tarefas são concorrentes, o download do arquivo2 termina antes do arquivo1, mesmo tendo sido criado depois. Isso demonstra o poder de não bloquear o programa durante operações de I/O.

É importante lembrar que await só pode ser usado dentro de funções definidas com async def. Tentar usar await fora de uma corrotina resultará em erro de sintaxe. Além disso, chamar uma corrotina sem await ou sem create_task não executa seu corpo: ela apenas cria um objeto que pode ser aguardado mais tarde.

Event loop

O event loop é o centro do asyncio: ele é responsável por gerenciar a execução de todas as corrotinas, decidindo qual delas deve rodar a cada momento. Quando uma corrotina encontra um await, ela é suspensa e o controle volta ao event loop, que pode executar outras tarefas prontas. Quando a operação aguardada é concluída (por exemplo, um timer expira ou uma resposta de rede chega), o event loop retoma a corrotina a partir do ponto onde ela parou.

Em versões anteriores ao Python 3.10, era comum usar funções como asyncio.get_event_loop() para obter o loop e loop.run_until_complete() para executar uma corrotina. Hoje, a maneira mais simples e recomendada é usar asyncio.run(), que cria um novo event loop, executa a corrotina dada e o fecha ao final. Vamos ver um exemplo que mostra o event loop em ação com várias tarefas:

import asyncio

async def tarefa(nome, atraso):
    print(f"Tarefa {nome} iniciou")
    await asyncio.sleep(atraso)
    print(f"Tarefa {nome} terminou após {atraso}s")

async def main():
    print("Início do main")
    tarefas = [
        asyncio.create_task(tarefa("A", 2)),
        asyncio.create_task(tarefa("B", 1)),
        asyncio.create_task(tarefa("C", 3)),
    ]
    # Aguarda todas as tarefas
    await asyncio.gather(*tarefas)
    print("Fim do main")

asyncio.run(main())

Observe que a ordem de término não é a mesma da criação: a tarefa B termina primeiro, depois A e C. O event loop gerencia a intercalação das tarefas conforme elas estão prontas. Você pode ver o event loop como um agendador que mantém uma fila de tarefas prontas e executa cada uma até que ela precise esperar.

Uma boa analogia é um chef de cozinha que, enquanto o arroz cozinha (espera), corta legumes (executa outra tarefa). O event loop faz o mesmo: em vez de ficar parado esperando uma operação de I/O, ele aproveita para executar outras corrotinas.

Quando usar

A programação assíncrona com asyncio é extremamente útil em aplicações que fazem muitas operações de I/O, como servidores web, clientes de APIs, scraping, aplicações de rede, etc. O principal benefício é a eficiência: você pode lidar com milhares de conexões simultâneas em um único thread, sem o overhead de criar threads ou processos.

No entanto, asyncio não é uma bala de prata. Para tarefas que são intensivas em CPU (como cálculos matemáticos pesados), a programação assíncrona não traz ganhos, pois o código que consome CPU não libera o event loop para outras tarefas. Nesses casos, o ideal é usar multiprocessamento (módulo multiprocessing) ou threads para paralelismo verdadeiro.

Também é importante notar que asyncio tem uma curva de aprendizado e pode tornar o código mais complexo. Para scripts simples que não exigem concorrência, usar código síncrono é mais simples e legível. A regra de ouro é: use asyncio quando você tiver muitas operações de I/O que podem ser executadas concorrentemente e quando a performance for um requisito importante.

Um exemplo típico de uso é um servidor web que precisa responder a várias requisições ao mesmo tempo. Com asyncio, você pode escrever um servidor que aceita conexões e processa cada uma de forma assíncrona, sem bloquear o aceite de novas conexões. Abaixo, um esboço de um servidor TCP simples:

import asyncio

async def handle_client(reader, writer):
    data = await reader.read(100)
    message = data.decode()
    print(f"Recebido: {message}")
    writer.write(data)  # Ecoa de volta
    await writer.drain()
    writer.close()

async def main():
    server = await asyncio.start_server(handle_client, '127.0.0.1', 8888)
    print("Servidor iniciado")
    async with server:
        await server.serve_forever()

asyncio.run(main())

Esse servidor pode atender múltiplos clientes simultaneamente, pois cada conexão é tratada em uma corrotina separada. Isso é um exemplo real de onde asyncio brilha.

Boas práticas e observações

Ao trabalhar com asyncio, algumas práticas ajudam a evitar erros comuns:

  • Sempre use asyncio.run() para iniciar o programa, em vez de criar e gerenciar o event loop manualmente.
  • Não chame funções bloqueantes (como time.sleep() ou requests.get()) dentro de corrotinas; use as versões assíncronas (asyncio.sleep(), aiohttp, etc.) ou execute-as em threads com asyncio.to_thread().
  • Use asyncio.gather() ou asyncio.create_task() para executar várias corrotinas concorrentemente, em vez de aguardá-las em sequência.
  • Tenha cuidado com o uso de variáveis globais compartilhadas: como o código é single-threaded, não há corrida de dados, mas pode haver interleaving entre await que cause comportamentos inesperados.
  • Prefira bibliotecas assíncronas da comunidade, como aiohttp para HTTP, asyncpg para PostgreSQL, etc., para obter o máximo de desempenho.

Exercícios

  1. Escreva uma corrotina que imprime os números de 1 a 5, com um asyncio.sleep(0.5) entre cada número. Execute-a com asyncio.run().

    ✓ Resposta: 
    import asyncio

async def contar():
    for i in range(1, 6):
        print(i)
        await asyncio.sleep(0.5)

asyncio.run(contar())

  2. Crie duas corrotinas que simulam tarefas de download com tempos diferentes. Use asyncio.gather() para executá-las concorrentemente e meça o tempo total de execução.

    ✓ Resposta: 
    import asyncio
import time

async def download(nome, tempo):
    print(f"Baixando {nome}...")
    await asyncio.sleep(tempo)
    print(f"{nome} concluído")
    return f"{nome} baixado"

async def main():
    inicio = time.time()
    resultados = await asyncio.gather(
        download("arquivo1", 3),
        download("arquivo2", 1)
    )
    print(f"Tempo total: {time.time() - inicio:.2f} s")
    print(resultados)

asyncio.run(main())

  3. Explique por que o código a seguir falha ao tentar usar await fora de uma corrotina e como corrigi-lo.

    import asyncio

async def rotina():
    print("Olá")

rotina()  # Isso não executa a rotina

    ✓ Resposta:

    O código não executa a rotina porque rotina() apenas retorna um objeto de corrotina, que não é aguardado. Para executar, você precisa usar asyncio.run(rotina()) ou criar uma tarefa dentro de outra corrotina com asyncio.create_task().

  4. Escreva um programa que usa asyncio.Queue para simular um produtor e um consumidor. O produtor coloca números de 1 a 5 na fila com um pequeno atraso, e o consumidor os retira e imprime.

    ✓ Resposta: 
    import asyncio

async def produtor(fila):
    for i in range(1, 6):
        await asyncio.sleep(0.2)
        await fila.put(i)
        print(f"Produzido: {i}")
    await fila.put(None)  # Sinal de fim

async def consumidor(fila):
    while True:
        item = await fila.get()
        if item is None:
            break
        print(f"Consumido: {item}")

async def main():
    fila = asyncio.Queue()
    await asyncio.gather(
        produtor(fila),
        consumidor(fila)
    )

asyncio.run(main())

  5. Qual é a diferença entre asyncio.create_task e simplesmente aguardar uma corrotina com await? Dê um exemplo que ilustre a diferença.

    ✓ Resposta:

    await corrotina() executa a corrotina imediatamente e bloqueia a corrotina atual até que ela termine. asyncio.create_task(corrotina()) agenda a corrotina para execução concorrente, permitindo que o código continue até que um await seja necessário. Exemplo:

    import asyncio

async def tarefa(nome):
    print(f"Início {nome}")
    await asyncio.sleep(1)
    print(f"Fim {nome}")

async def main():
    # Com await: executa sequencialmente
    await tarefa("A")
    await tarefa("B")

    # Com create_task: concorrente
    t1 = asyncio.create_task(tarefa("C"))
    t2 = asyncio.create_task(tarefa("D"))
    await t1
    await t2

asyncio.run(main())

    No primeiro caso, A termina antes de B começar. No segundo, C e D são executados concorrentemente, e o tempo total é menor.

Referências