Introdução à Concorrência Eficiente

A programação assíncrona emerge como uma solução robusta para aplicações intensivas em I/O, como operações de rede, manipulação de arquivos e interações com bancos de dados. Diferente do modelo tradicional bloqueante, onde o processo pausa até a conclusão de uma operação, o modelo assíncrono permite que o fluxo de execução seja liberado durante períodos de espera. Isso possibilita que outras tarefas sejam processadas, otimizando o uso de recursos e aumentando a throughput do sistema.

Nativo no Python a partir da versão 3.5, esse modelo utiliza as palavras-chave async e await para gerenciar corrotinas. O mecanismo subjacente opera dentro de uma única thread, utilizando um loop de eventos para alternar entre tarefas suspensas e prontas para execução, simulando concorrência sem a sobrecarga de múltiplas threads.

Fundamentos das Corrotinas em Python

O termo "corrotina" no ecossistema Python possui duas definições distintas dependendo do contexto:

1. Definição Sintática: Qualquer função declarada com async def é considerada uma corrotina. O corpo desse函数 pode conter expressões await. Tentar usar await dentro de uma função padrão (def) resulta em erro de sintaxe.
2. Objeto em Tempo de Execução: Ao invocar uma função definida com async def, o retorno não é o resultado da função, mas sim um objeto corrotina. Este objeto deve ser agendado no loop de eventos ou passado para outra corrotina via await.

Na prática, a unidade real de execução gerenciada pelo loop de eventos é a Tarefa (Task). Quando uma corrotina é aguardada (await), o loop assume o controle para调度 outras tarefas pendentes até que a operação original seja finalizada.

Componentes Principais

• Event Loop (Loop de Eventos): O núcleo do sistema assíncrono. Responsável por registrar, agendar e executar corrotinas.
• Coroutine (Corrotina): Função capaz de suspender sua execução em pontos específicos e retomá-la posteriormente.
• Task (Tarefa): Um wrapper que envolve uma corrotina para gerenciar seu estado e execução dentro do loop.

O operador await no módulo asyncio opera de duas formas principais: aguardando diretamente uma corrotina (execução sequencial) ou aguardando uma Task (permitindo concorrência enquanto o loop processa outras pendências).

Sintaxe e Palavras-Chave

Definindo Funções Assíncronas

O uso do modificador async antes de def transforma uma função comum em uma corrotina. É crucial notar que a chamada dessa função não executa seu código imediatamente; ela retorna um objeto aguardável.

async def gerar_relatorio_sistema():
    return "Status: Operacional"

# A chamada retorna o objeto, não executa o corpo
objeto_corrotina = gerar_relatorio_sistema()
print(type(objeto_corrotina))  # <class 'coroutine'>

Suspensão com Await

A palavra-chave await indica um ponto de pausa. O controle é devolvido ao loop de eventos, que pode executar outras tarefas. A execução só retoma quando o objeto aguardável finaliza.

import asyncio

async def processar_item(tempo, identificador):
    await asyncio.sleep(tempo)
    print(f"Item {identificador} processado")

async def fluxo_principal():
    print("Início do fluxo")
    await processar_item(1, 'A')
    await processar_item(1, 'B')
    print("Fim do fluxo")

Gerenciamento do Loop de Eventos

O loop de eventos é o motor que impulsiona a concorrência. Embora seja possível manipulá-lo manualmente, a abordagem recomendada para iniciar aplicações é utilizar asyncio.run(), que inicializa o loop, executa a corrotina principle e finaliza os recursos adequadamente.

import asyncio

async def main():
    # Lógica da aplicação
    pass

# Inicialização segura (Python 3.7+)
asyncio.run(main())

Execução Concorrente de Tarefas

O verdadeiro potencial do assincronismo aparece quando múltiplas operações são executadas simultaneamente. Isso é alcançado convertendo corrotinas em Tasks.

Criando Tasks com create_task

A função asyncio.create_task() agenda uma corrotina para execução imediata no background. O objeto Task retornado pode ser aguardado posteriormente.

import asyncio

async def executar_tarefa(tempo, msg):
    await asyncio.sleep(tempo)
    print(msg)

async def principal():
    t1 = asyncio.create_task(executar_tarefa(1, 'Processo 1'))
    t2 = asyncio.create_task(executar_tarefa(1, 'Processo 2'))
    
    print("Tarefas agendadas")
    
    await t1
    await t2
    print("Todas concluídas")

asyncio.run(principal())

Agrupamento com gather

Para esperar por múltiplos objetos aguardáveis simultaneamente e coletar seus retornos, asyncio.gather() é a ferramenta ideal. Ele preserva a ordem dos resultados baseada na ordem de entrada.

import asyncio

async def buscar_info(identificacao, duracao):
    print(f"Iniciando {identificacao}")
    await asyncio.sleep(duracao)
    print(f"Completo {identificacao}")
    return f"Dado {identificacao}"

async def orchestrator():
    resultados = await asyncio.gather(
        buscar_info(1, 2),
        buscar_info(2, 1),
        buscar_info(3, 3),
    )
    print("Resultados:", resultados)

asyncio.run(orchestrator())

Objetos Aguardáveis (Awaitables)

Três tipos principais podem ser usados com await:

1. Corrotinas: Retornadas por funções async def.
2. Tasks: Subclasses de Future, agendadas explicitamente.
3. Futures: Objetos de baixo nível representando um resultado futuro. Geralmente usados internamente por bibliotecas.

Práticas Recomendadas e Armadilhas Comuns

• Evite Bloqueio no Loop: Nunca use funções síncronas bloqueantes (ex: time.sleep, requests) dentro de corrotinas. Isso congela o loop de eventos. Utilize versões assíncronas (asyncio.sleep, aiohttp).
• Await é Obrigatório: Se uma corrotina ou Task não for aguardada, ela pode não executar ou seus erros podem ser silenciosamente ignorados.
• Concorrência vs Paralelismo: O asyncio é concorrente, não paralelo. Ele opera em uma única thread. Para tarefas intensivas de CPU, utilize o módulo multiprocessing.
• Contextos Assíncronos: Utilize async with para gerenciar recursos que suportam protocolo de contexto assíncrono e async for para iteráveis assíncronos.

Cenários de Uso Real

Requisições HTTP Concrorentes

Utilizando aiohttp, é possível realizar múltiplas chamadas de API sem bloquear a thread principle.

import aiohttp
import asyncio
import time

async def obter_dados(sessao, url, atraso=1):
    inicio = time.time()
    async with sessao.get(f'https://httpbin.org/delay/{atraso}') as resp:
        await resp.json()
        decorrido = time.time() - inicio
        print(f"{url} finalizado em {decorrido:.2f}s")

async def execucao_sequencial():
    async with aiohttp.ClientSession() as sessao:
        start = time.time()
        for i in range(3):
            await obter_dados(sessao, f'req_{i}', 1)
        print(f"Sequencial: {time.time() - start:.2f}s")

async def execucao_paralela():
    async with aiohttp.ClientSession() as sessao:
        start = time.time()
        tarefas = [
            obter_dados(sessao, f'par_{i}', 1) for i in range(5)
        ]
        await asyncio.gather(*tarefas)
        print(f"Paralelo: {time.time() - start:.2f}s")

async def main():
    await execucao_sequencial()
    await execucao_paralela()

asyncio.run(main())

Operações de Arquivo com aiofiles

Para I/O de disco, a biblioteca aiofiles permite leitura e escrita não bloqueantes.

import aiofiles
import asyncio
import os

async def salvar_registro(nome, conteudo):
    async with aiofiles.open(nome, 'w', encoding='utf-8') as arq:
        await arq.write(conteudo)
    print(f"Escrito: {nome}")

async def ler_registro(nome):
    async with aiofiles.open(nome, 'r', encoding='utf-8') as arq:
        dados = await arq.read()
    print(f"Lido: {nome} ({len(dados)} chars)")
    return dados

async def gerenciar_arquivos():
    arquivos = {
        'log_1.txt': 'Dados do sistema 1' * 500,
        'log_2.txt': 'Dados do sistema 2' * 500,
        'log_3.txt': 'Dados do sistema 3' * 500,
    }
    
    tarefas_escrita = [salvar_registro(k, v) for k, v in arquivos.items()]
    await asyncio.gather(*tarefas_escrita)
    
    tarefas_leitura = [ler_registro(k) for k in arquivos.keys()]
    await asyncio.gather(*tarefas_leitura)
    
    for k in arquivos.keys():
        os.remove(k)

asyncio.run(gerenciar_arquivos())

Consultas a Banco de Dados com asyncpg

O driver asyncpg para PostgreSQL permite pool de conexões e queries assíncronas.

import asyncio
import asyncpg

async def configurar_pool():
    return await asyncpg.create_pool(
        user='usuario',
        password='senha',
        database='teste',
        host='localhost',
        min_size=5,
        max_size=10
    )

async def executar_query(conn, id_query):
    import time
    inicio = time.time()
    resultado = await conn.fetchval("SELECT COUNT(*) FROM produtos")
    await asyncio.sleep(0.1) # Simulação de processamento
    print(f"Query {id_query}: {resultado} em {time.time() - inicio:.2f}s")
    return resultado

async def rodar_queries():
    pool = await configurar_pool()
    async with pool.acquire() as conn:
        tarefas = [executar_query(conn, i) for i in range(5)]
        await asyncio.gather(*tarefas)
    await pool.close()

# asyncio.run(rodar_queries())