Entendendo Async/Await: Fundamentos e Diferença do Síncrono
Quando começamos a programar, aprendemos a escrever código sequencial: uma linha executa, depois a próxima, e assim por diante. Isso funciona bem para a maioria dos casos, mas há situações onde essa abordagem cria gargalos significativos. Imagine fazer múltiplas requisições HTTP ou operações de leitura de arquivo — seu programa ficaria parado esperando cada uma terminar antes de começar a próxima.
A programação assíncrona resolve esse problema permitindo que seu código comece uma operação, a deixe "em progresso" e continue executando outras tarefas. Quando a operação termina, você retoma o trabalho. Em Python, async e await são as palavras-chave que tornam isso possível. async define uma função assíncrona, enquanto await pausa a execução até que um resultado chegue, sem bloquear o programa inteiro.
# Código síncrono (bloqueante)
import time
def buscar_dados(id):
time.sleep(2) # Simula I/O
return f"Dados do usuário {id}"
def processar_usuarios():
for i in range(3):
resultado = buscar_dados(i)
print(resultado)
inicio = time.time()
processar_usuarios()
print(f"Tempo total: {time.time() - inicio:.2f}s") # ~6 segundos
Agora, veja como async/await muda isso:
# Código assíncrono (não-bloqueante)
import asyncio
import time
async def buscar_dados(id):
await asyncio.sleep(2) # Simula I/O sem bloquear
return f"Dados do usuário {id}"
async def processar_usuarios():
tarefas = [buscar_dados(i) for i in range(3)]
resultados = await asyncio.gather(*tarefas)
for resultado in resultados:
print(resultado)
inicio = time.time()
asyncio.run(processar_usuarios())
print(f"Tempo total: {time.time() - inicio:.2f}s") # ~2 segundos
A diferença é clara: no primeiro caso, levamos 6 segundos porque cada operação espera a anterior terminar. No segundo, levamos apenas 2 segundos porque as três operações rodam concorrentemente. Isso não é paralelismo real (threads ou processos), é o gerenciador de eventos (event loop) do Python alternando entre tarefas de forma eficiente.
Padrões Essenciais: Como Estruturar Código Assíncrono
O Event Loop: O Coração da Execução Assíncrona
O event loop é um mecanismo que controla a execução de corrotinas no Python. Ele mantém uma fila de tarefas prontas para executar, e quando uma pausa (com await), o loop executa outra. Você raramente cria um event loop manualmente; asyncio.run() faz isso para você em scripts simples, mas é importante entender seu papel.
import asyncio
async def tarefa_a():
print("Tarefa A iniciada")
await asyncio.sleep(1)
print("Tarefa A concluída")
async def tarefa_b():
print("Tarefa B iniciada")
await asyncio.sleep(1)
print("Tarefa B concluída")
async def main():
# Executa A e B concorrentemente
await asyncio.gather(tarefa_a(), tarefa_b())
asyncio.run(main())
Gather: Executando Múltiplas Corrotinas
asyncio.gather() é o seu aliado para executar várias corrotinas simultaneamente. Ele retorna uma lista com todos os resultados na mesma ordem das corrotinas passadas. Se uma corrotina lançar exceção, por padrão toda a operação falha — mas você pode controlar isso com return_exceptions=True.
async def buscar_usuario(id):
await asyncio.sleep(1)
if id == 2:
raise ValueError(f"Usuário {id} não encontrado")
return f"Usuário {id}"
async def main():
# Sem tratamento de erro
try:
resultados = await asyncio.gather(
buscar_usuario(1),
buscar_usuario(2),
buscar_usuario(3)
)
except ValueError as e:
print(f"Erro capturado: {e}")
# Com tratamento seguro
resultados = await asyncio.gather(
buscar_usuario(1),
buscar_usuario(2),
buscar_usuario(3),
return_exceptions=True
)
for i, resultado in enumerate(resultados):
if isinstance(resultado, Exception):
print(f"Erro no índice {i}: {resultado}")
else:
print(f"Sucesso: {resultado}")
asyncio.run(main())
Create Task: Agendando Tarefas com Mais Controle
Enquanto gather() é conveniente, asyncio.create_task() oferece mais controle. Ele cria uma tarefa e a agenda para execução, retornando imediatamente um objeto Task que você pode aguardar depois. Isso é útil quando você precisa iniciar múltiplas operações sem esperar que uma termine antes de iniciar a próxima.
import asyncio
async def processar_pedido(id_pedido):
print(f"Processando pedido {id_pedido}")
await asyncio.sleep(2)
print(f"Pedido {id_pedido} concluído")
return f"Resultado do pedido {id_pedido}"
async def main():
# Cria tarefas sem aguardar
tarefa1 = asyncio.create_task(processar_pedido(1))
tarefa2 = asyncio.create_task(processar_pedido(2))
# Faz algo enquanto elas rodam
print("Tarefas iniciadas, fazendo outra coisa...")
await asyncio.sleep(0.5)
# Aguarda ambas
resultado1 = await tarefa1
resultado2 = await tarefa2
print(resultado1, resultado2)
asyncio.run(main())
Armadilhas Comuns e Como Evitá-las
Armadilha 1: Esquecer de await
A armadilha mais clássica é chamar uma função assíncrona sem await. Você não receberá um erro imediato — apenas obterá um objeto corrotina não iniciado. Seu código continuará rodando, mas a operação nunca será executada. Isso deixa muitos iniciantes confusos.
# ❌ ERRADO
async def buscar_dados():
await asyncio.sleep(1)
return "dados"
async def main():
resultado = buscar_dados() # Sem await!
print(resultado) # <coroutine object buscar_dados at 0x...>
# Aviso não suprimido!
asyncio.run(main())
# ✅ CORRETO
async def buscar_dados():
await asyncio.sleep(1)
return "dados"
async def main():
resultado = await buscar_dados() # Com await
print(resultado) # "dados"
asyncio.run(main())
Armadilha 2: Misturar Código Síncrono e Assíncrono
Você não pode chamar await dentro de uma função normal (não assíncrona). Da mesma forma, operações bloqueantes (como time.sleep() ou requisições síncronas) travamtodo o event loop se executadas diretamente. Use bibliotecas assíncronas específicas.
# ❌ ERRADO: operação bloqueante em código assíncrono
import time
async def main():
time.sleep(2) # Bloqueia todo o event loop!
print("Finalmente")
asyncio.run(main())
# ✅ CORRETO: use equivalentes assíncrono
async def main():
await asyncio.sleep(2) # Não bloqueia
print("Finalmente")
asyncio.run(main())
Se você realmente precisa executar código bloqueante, use asyncio.to_thread() (Python 3.9+) ou loop.run_in_executor():
import asyncio
import time
def tarefa_bloqueante():
time.sleep(2)
return "Pronto"
async def main():
# Executa em uma thread separada, sem bloquear o event loop
resultado = await asyncio.to_thread(tarefa_bloqueante)
print(resultado)
asyncio.run(main())
Armadilha 3: Exceções Silenciosas em Tarefas
Quando você cria uma tarefa com create_task(), as exceções não propagam automaticamente. Se você nunca aguardar a tarefa, o erro passará despercebido até o garbage collector limpar a tarefa, momento em que Python imprimirá um aviso feio. Sempre aguarde suas tarefas ou use callbacks.
# ❌ ERRADO: exceção silenciosa
async def operacao_com_erro():
await asyncio.sleep(1)
raise ValueError("Algo deu errado")
async def main():
# A tarefa é criada mas o erro nunca é tratado
tarefa = asyncio.create_task(operacao_com_erro())
print("Continuando...")
# Se main() terminar sem aguardar tarefa, veremos um aviso feio
asyncio.run(main())
# ✅ CORRETO: sempre aguarde ou use done_callback
async def operacao_com_erro():
await asyncio.sleep(1)
raise ValueError("Algo deu errado")
async def main():
tarefa = asyncio.create_task(operacao_com_erro())
try:
resultado = await tarefa
except ValueError as e:
print(f"Erro capturado: {e}")
asyncio.run(main())
Armadilha 4: Deadlocks com Locks
Usar asyncio.Lock() incorretamente pode criar deadlocks. A armadilha comum é esquecer de usar async with ou aguardar a aquisição do lock.
# ❌ ERRADO: pode causar deadlock
import asyncio
lock = asyncio.Lock()
async def tarefa():
await lock.acquire()
# Se algum erro ocorrer aqui, lock nunca é liberado
await asyncio.sleep(1)
lock.release()
async def main():
await asyncio.gather(tarefa(), tarefa()) # Deadlock!
asyncio.run(main())
# ✅ CORRETO: use async with
async def tarefa():
async with lock:
await asyncio.sleep(1)
print("Executando")
async def main():
await asyncio.gather(tarefa(), tarefa())
asyncio.run(main())
Armadilha 5: Assumir que Async é Mais Rápido Sempre
Async brilha em operações I/O (rede, arquivos). Para computação CPU-intensiva, não ajuda — de fato, pode ser mais lento devido ao overhead. Use asyncio.to_thread() ou multiprocessing para CPU.
import asyncio
import time
# CPU-intensiva
def fibonacci(n):
if n < 2:
return n
return fibonacci(n-1) + fibonacci(n-2)
async def calcular():
# Isso é ineficiente para CPU-intensiva
resultado = fibonacci(35)
return resultado
inicio = time.time()
asyncio.run(calcular())
print(f"Tempo: {time.time() - inicio:.2f}s")
# Melhor: use threads para isso
async def calcular_com_thread():
resultado = await asyncio.to_thread(fibonacci, 35)
return resultado
inicio = time.time()
asyncio.run(calcular_com_thread())
print(f"Tempo: {time.time() - inicio:.2f}s")
Padrões Avançados: Timeout, Streaming e Contexto
Timeout com asyncio.wait_for()
Operações assíncronas precisam de proteção contra travamentos infinitos. asyncio.wait_for() define um prazo máximo de espera.
import asyncio
async def operacao_lenta():
await asyncio.sleep(5)
return "Concluído"
async def main():
try:
resultado = await asyncio.wait_for(operacao_lenta(), timeout=2)
print(resultado)
except asyncio.TimeoutError:
print("Operação excedeu o tempo limite")
asyncio.run(main())
Streaming de Dados com Filas
Para processar grandes volumes de dados, asyncio.Queue oferece um padrão produtor-consumidor eficiente.
import asyncio
async def produtor(fila):
for i in range(5):
await asyncio.sleep(1)
await fila.put(f"Item {i}")
print(f"Produzido: Item {i}")
async def consumidor(fila):
while True:
item = await fila.get()
if item is None:
break
print(f"Consumido: {item}")
await asyncio.sleep(0.5)
fila.task_done()
async def main():
fila = asyncio.Queue()
# Executa produtor e consumidor
await asyncio.gather(
produtor(fila),
consumidor(fila)
)
# Sinaliza fim
await fila.put(None)
asyncio.run(main())
Context Managers Assíncrono
Para recursos que precisam de setup e cleanup (conexões, transações), use async with.
import asyncio
class ConexaoDB:
async def __aenter__(self):
print("Conectando ao banco...")
await asyncio.sleep(1)
return self
async def __aexit__(self, exc_type, exc_val, exc_tb):
print("Desconectando do banco...")
await asyncio.sleep(1)
async def consulta(conexao):
print("Executando consulta...")
await asyncio.sleep(1)
return "Resultado"
async def main():
async with ConexaoDB() as db:
resultado = await consulta(db)
print(resultado)
asyncio.run(main())
Conclusão
Nesta jornada, você aprendeu que async/await não é paralelismo, mas concorrência eficiente para operações I/O. O event loop é o maestro que alterna entre tarefas quando uma pausa com await, permitindo que múltiplas operações rodem sem bloquear a execução. Em segundo lugar, você identificou as armadilhas mais perigosas: esquecer de await, misturar código bloqueante com assíncrono, deixar tarefas com exceções silenciosas, criar deadlocks com locks e aplicar async onde não é apropriado. Por fim, entendeu que o real poder vem de padrões bem estruturados como gather(), create_task(), timeouts, filas e context managers, que transformam código assíncrono complexo em soluções elegantes e eficientes.