0Pricing
Python Academy · Aula

Padrões de E/S assíncrona em produção

Aplique asyncio a clientes HTTP, filas e pipelines do mundo real.

Padrões de E/S assíncrona em produção é uma aula grátis de Python Academy no CoddyKit. Esta é a aula 4 de 4. Você pode ler a aula completa abaixo gratuitamente — depois pratica ao vivo no navegador com um editor de código integrado e um tutor de IA 24/7. Faz parte do caminho de aprendizado de Python Academy, e seu progresso é sincronizado entre a web e o app CoddyKit. O curso de Python Academy inclui 4 aulas no total.

Práticas recomendadas para o ciclo de eventos

Sempre use asyncio.run() como o único ponto de entrada. Evite misturar código síncrono e assíncrono de forma descuidada e nunca bloqueie o ciclo de eventos com trabalho longo de CPU.

import asyncio

async def main():
    # Never: time.sleep() — blocks the event loop
    # Always: await asyncio.sleep()
    await asyncio.sleep(1)

asyncio.run(main())

Reutilização de conexões

Reutilize as conexões em vez de abrir uma nova para cada solicitação. Bibliotecas como asyncpg e aiohttp fornecem conjuntos de conexões integrados.

# pip install asyncpg
import asyncpg, asyncio

async def main():
    pool = await asyncpg.create_pool(
        dsn="postgresql://user:pass@host/db",
        min_size=5, max_size=20
    )
    async with pool.acquire() as conn:
        row = await conn.fetchrow("SELECT 1")
    await pool.close()

Produtor-consumidor com asyncio.Queue

asyncio.Queue desacopla produtores e consumidores executados como tarefas simultâneas.

import asyncio

async def producer(q):
    for i in range(5):
        await q.put(i)
        await asyncio.sleep(0.5)
    await q.put(None)  # sentinel

async def consumer(q):
    while (item := await q.get()) is not None:
        print(f"processing {item}")
        q.task_done()

async def main():
    q = asyncio.Queue()
    await asyncio.gather(producer(q), consumer(q))

asyncio.run(main())

Limitação de taxa com Semaphore

Limite as solicitações simultâneas de saída para evitar sobrecarregar APIs externas ou atingir limites de taxa.

import asyncio, aiohttp

SEM = asyncio.Semaphore(10)  # 10 concurrent requests max

async def fetch(session, url):
    async with SEM:
        async with session.get(url) as r:
            return await r.json()

async def main(urls):
    async with aiohttp.ClientSession() as s:
        return await asyncio.gather(*[fetch(s, u) for u in urls])

Nova tentativa com espera exponencial

Envolva operações que podem ser repetidas com uma lógica de espera progressiva para lidar adequadamente com falhas temporárias.

import asyncio

async def retry(coro_fn, retries=3, base=1):
    for attempt in range(retries):
        try:
            return await coro_fn()
        except Exception as e:
            if attempt == retries - 1: raise
            delay = base * 2 ** attempt
            await asyncio.sleep(delay)

Desligamento normal

Capture asyncio.CancelledError em tarefas de longa duração e trate os sinais do sistema operacional para desligar de forma limpa.

import asyncio, signal

async def server():
    stop = asyncio.Event()
    loop = asyncio.get_running_loop()
    loop.add_signal_handler(signal.SIGINT, stop.set)
    await stop.wait()
    print("Shutting down gracefully")

Respostas em fluxo

Use geradores assíncronos para transmitir respostas grandes sem carregá-las inteiramente na memória.

import asyncio, aiohttp

async def stream(url):
    async with aiohttp.ClientSession() as s:
        async with s.get(url) as r:
            async for chunk in r.content.iter_chunked(4096):
                process(chunk)

async def process(chunk):
    await asyncio.sleep(0)
    print(len(chunk), "bytes")

Tarefas em segundo plano no FastAPI

O BackgroundTasks do FastAPI enfileira o trabalho para ser executado depois que a resposta é enviada, mantendo os pontos de acesso rápidos.

from fastapi import FastAPI, BackgroundTasks

app = FastAPI()

def send_email(to: str, body: str):
    # heavy synchronous work
    pass

@app.post("/register")
async def register(email: str, bg: BackgroundTasks):
    bg.add_task(send_email, email, "Welcome!")
    return {"status": "registered"}

Padrão de sinal de atividade

Execute uma tarefa periódica em segundo plano junto ao ciclo principal do servidor usando create_task.

import asyncio

async def heartbeat():
    while True:
        print("alive")
        await asyncio.sleep(30)

async def main():
    asyncio.create_task(heartbeat())
    await serve_forever()

Criação de perfil de código assíncrono

Use yappi ou py-spy para criar perfis conscientes de operações assíncronas. Ative o modo de depuração do asyncio para receber avisos sobre retornos de chamada lentos.

# Enable debug to catch slow callbacks:
# asyncio.run(main(), debug=True)

# PYTHONASYNCIODEBUG=1 python server.py
# logs: Executing <Task coroutine=...> took 0.150 seconds

Evitando armadilhas comuns

Regras principais: nunca chame código bloqueante no ciclo de eventos; sempre aguarde corrotinas; cancele tarefas durante o desligamento; use conjuntos de conexões; limite a simultaneidade com semáforos.

import asyncio

async def bad():
    import time
    time.sleep(2)        # blocks the entire event loop!

async def good():
    await asyncio.sleep(2)  # yields control

Verificação rápida

Qual primitiva do asyncio desacopla produtores e consumidores executados como tarefas simultâneas?

Recapitulação

Padrões assíncronos para produção: use conjuntos de conexões, controle a simultaneidade com Semaphore, implemente novas tentativas com espera progressiva, transmita dados grandes, desligue normalmente ao receber sinais e execute tarefas periódicas com create_task.

Perguntas Frequentes

A aula “Padrões de E/S assíncrona em produção” é grátis?

Sim — o texto completo de “Padrões de E/S assíncrona em produção” é grátis para ler aqui na web. Para praticá-la interativamente (um editor de código integrado e um tutor de IA 24/7) e desbloquear o restante do curso de Python Academy, atualize para CoddyKit PRO. O curso de Python Academy inclui 4 aulas no total.

O que vou aprender em “Padrões de E/S assíncrona em produção”?

Aplique asyncio a clientes HTTP, filas e pipelines do mundo real. Você pratica Python Academy com código prático que executa diretamente no navegador, e um tutor de IA 24/7 responde suas dúvidas enquanto trabalha na aula.

Preciso ter experiência prévia para começar Python Academy?

Nenhuma experiência prévia é necessária. Python Academy no CoddyKit é estruturado para alunos iniciantes até avançados, então você pode começar aqui ou desde o início e aprender no seu ritmo. Esta é a aula 4 de 4.

Quanto tempo leva a aula “Padrões de E/S assíncrona em produção”?

A maioria das aulas CoddyKit leva cerca de 5–10 minutos. Cada uma é compacta e interativa, então você faz progresso constante e retoma exatamente de onde parou entre web e app.

Posso escrever e executar código nesta aula de Python Academy?

Sim. Cada aula de Python Academy inclui um editor de código integrado, então você escreve e executa código real direto no navegador e recebe feedback de IA instantaneamente — nenhuma configuração local necessária.

Todas as aulas deste curso

  1. Corrotinas e o laço de eventos
  2. await, tarefas e agrupamento
  3. Gerenciadores de contexto e iteradores assíncronos
  4. Padrões de E/S assíncrona em produção
← Voltar para Python Academy