A programação assíncrona é uma das técnicas mais poderosas para otimizar o desempenho de aplicações Python, especialmente quando lidamos com operações que envolvem espera, como requisições de rede, leitura de arquivos ou consultas a bancos de dados. O asyncio é a biblioteca nativa do Python que torna possível escrever código assíncrono de forma eficiente e elegante.
Neste guia completo, você vai aprender desde os conceitos fundamentais até exemplos práticos de como usar o asyncio para criar aplicações mais rápidas e responsivas. Vamos explorar corrotinas, event loops, tasks e as palavras-chave async e await que revolucionaram a forma como escrevemos código concorrente em Python.
Para complementar este conteúdo, confira este excelente vídeo da Hashtag Programação que explica programação assíncrona em Python de forma didática:
Créditos: Hashtag Programação.
O que é Programação Assíncrona
Antes de mergulharmos no asyncio, é essencial entender o conceito de programação assíncrona. Na programação tradicional, o código é executado de forma síncrona, ou seja, linha por linha, esperando cada operação terminar antes de iniciar a próxima.
Imagine que você está preparando um café. Na abordagem síncrona, você ferveria a água, esperaria terminar completamente, só então pegaria o pó de café, esperaria novamente, e assim por diante. Esse processo é extremamente ineficiente quando há tarefas que naturalmente envolvem tempo de espera.
A programação assíncrona permite que múltiplas tarefas sejam executadas de forma concorrente. Enquanto a água está fervendo, você pode pegar o pó de café, preparar a xícara e realizar outras atividades. O programa não fica bloqueado esperando uma única operação terminar.
É importante destacar que programação assíncrona não é o mesmo que multithreading ou multiprocessing. O asyncio trabalha com um único thread, alternando entre tarefas de forma inteligente sempre que uma delas precisa esperar por algo, como uma resposta de rede ou leitura de disco.
O que é o Asyncio no Python
O asyncio é uma biblioteca padrão do Python, introduzida na versão 3.4 e significativamente melhorada na versão 3.7. Ela fornece toda a infraestrutura necessária para escrever código assíncrono usando as palavras-chave async e await.
Esta biblioteca é especialmente útil para operações I/O-bound, ou seja, operações limitadas por entrada e saída de dados. Alguns exemplos práticos incluem:
- Fazer múltiplas requisições HTTP simultaneamente
- Ler e escrever arquivos de forma não bloqueante
- Gerenciar conexões de rede e websockets
- Consultar bancos de dados de forma assíncrona
- Criar servidores web de alto desempenho
O asyncio já vem instalado com o Python, então você não precisa instalar nenhuma biblioteca adicional. A partir da versão 3.7, a sintaxe foi simplificada, tornando o uso ainda mais intuitivo para desenvolvedores de todos os níveis.
Conceitos Fundamentais do Asyncio
Corrotinas (Coroutines)
Uma corrotina é uma função especial que pode ser pausada e retomada. Diferente das funções normais em Python, que executam do início ao fim sem interrupções, as corrotinas podem “esperar” por operações assíncronas sem bloquear o programa.
Para criar uma corrotina, você usa a palavra-chave async def em vez de apenas def:
import asyncio
async def minha_corrotina():
print("Início da corrotina")
await asyncio.sleep(1)
print("Fim da corrotina")
return "Resultado"Quando você chama uma função assíncrona, ela não executa imediatamente. Em vez disso, ela retorna um objeto corrotina que precisa ser aguardado (await) ou executado pelo event loop.
Event Loop
O event loop é o coração do asyncio. Ele funciona como um maestro de orquestra, gerenciando quando cada corrotina deve ser executada, pausada ou retomada. O event loop monitora constantemente quais tarefas estão prontas para serem executadas e alterna entre elas de forma eficiente.
Na maioria dos casos, você não precisa interagir diretamente com o event loop. A função asyncio.run() cuida de criar, executar e encerrar o event loop automaticamente:
import asyncio
async def main():
print("Executando no event loop")
await asyncio.sleep(1)
print("Concluído")
# Executa a corrotina principal
asyncio.run(main())Tasks (Tarefas)
Uma task é uma corrotina empacotada que foi agendada para execução no event loop. Tasks permitem que múltiplas corrotinas sejam executadas concorrentemente. Você cria tasks usando asyncio.create_task():
import asyncio
async def tarefa1():
await asyncio.sleep(2)
print("Tarefa 1 concluída")
async def tarefa2():
await asyncio.sleep(1)
print("Tarefa 2 concluída")
async def main():
# Cria e agenda as tasks
task1 = asyncio.create_task(tarefa1())
task2 = asyncio.create_task(tarefa2())
# Aguarda ambas concluírem
await task1
await task2
asyncio.run(main())No exemplo acima, ambas as tarefas são executadas concorrentemente. A tarefa 2 terminará primeiro (após 1 segundo), mesmo tendo sido criada depois da tarefa 1.
Async e Await
As palavras-chave async e await são a base da sintaxe assíncrona moderna em Python. A palavra async define uma função como assíncrona, enquanto await pausa a execução da corrotina até que a operação aguardada seja concluída.
É crucial entender que await só pode ser usado dentro de funções definidas com async def. Tentar usar await em uma função normal resultará em erro de sintaxe.
import asyncio
async def buscar_dados():
print("Buscando dados...")
await asyncio.sleep(2) # Simula operação demorada
print("Dados recebidos")
return {"nome": "Python", "versao": "3.11"}
async def processar():
dados = await buscar_dados() # Espera os dados
print(f"Processando: {dados}")
asyncio.run(processar())Como Instalar e Configurar o Asyncio
Uma das grandes vantagens do asyncio é que ele já vem incluído na instalação padrão do Python a partir da versão 3.4. Portanto, se você tem o Python 3.7 ou superior instalado, não precisa fazer nenhuma instalação adicional.
Para verificar se sua versão do Python suporta asyncio adequadamente, execute este simples teste:
import sys
import asyncio
print(f"Versão do Python: {sys.version}")
print(f"Versão do asyncio disponível: {asyncio.__version__ if hasattr(asyncio, '__version__') else 'Incluído na stdlib'}")Se você ainda está usando uma versão antiga do Python, recomendo fortemente atualizar para a versão mais recente. As melhorias no asyncio entre as versões 3.7 e 3.11+ são significativas, tornando o código mais simples e performático.
Primeiro Exemplo Prático com Asyncio
Vamos começar com um exemplo simples que demonstra a diferença entre código síncrono e assíncrono. Imagine que você precisa fazer download de dados de três URLs diferentes:
Versão Síncrona (Bloqueante)
import time
def download_dados(url, tempo):
print(f"Baixando de {url}...")
time.sleep(tempo) # Simula download
print(f"Concluído {url}")
return f"Dados de {url}"
def main():
inicio = time.time()
resultado1 = download_dados("url1.com", 2)
resultado2 = download_dados("url2.com", 2)
resultado3 = download_dados("url3.com", 2)
fim = time.time()
print(f"Tempo total: {fim - inicio:.2f} segundos")
main()
# Tempo total: 6 segundosNo código síncrono, cada download espera o anterior terminar. O tempo total é de 6 segundos (2+2+2).
Versão Assíncrona com Asyncio
import asyncio
import time
async def download_dados(url, tempo):
print(f"Baixando de {url}...")
await asyncio.sleep(tempo) # Simula download assíncrono
print(f"Concluído {url}")
return f"Dados de {url}"
async def main():
inicio = time.time()
# Cria tasks para execução concorrente
task1 = asyncio.create_task(download_dados("url1.com", 2))
task2 = asyncio.create_task(download_dados("url2.com", 2))
task3 = asyncio.create_task(download_dados("url3.com", 2))
# Aguarda todas as tasks
resultado1 = await task1
resultado2 = await task2
resultado3 = await task3
fim = time.time()
print(f"Tempo total: {fim - inicio:.2f} segundos")
asyncio.run(main())
# Tempo total: 2 segundosNa versão assíncrona, os três downloads acontecem simultaneamente. O tempo total é de apenas 2 segundos, uma redução de 67% no tempo de execução! Essa é a verdadeira força da programação assíncrona.
Trabalhando com Múltiplas Tarefas Assíncronas
O asyncio oferece várias formas de trabalhar com múltiplas tarefas. Vamos explorar as mais úteis:
asyncio.gather()
A função asyncio.gather() permite executar múltiplas corrotinas concorrentemente e coletar seus resultados em uma lista:
import asyncio
async def processar_item(numero):
await asyncio.sleep(1)
return numero * 2
async def main():
# Executa múltiplas corrotinas e coleta os resultados
resultados = await asyncio.gather(
processar_item(1),
processar_item(2),
processar_item(3),
processar_item(4)
)
print(f"Resultados: {resultados}")
# Saída: Resultados: [2, 4, 6, 8]
asyncio.run(main())O gather() espera todas as corrotinas terminarem e retorna os resultados na mesma ordem em que foram passadas, independente da ordem de conclusão.
asyncio.wait()
A função asyncio.wait() oferece mais controle sobre como esperar pelas tarefas. Você pode especificar se quer esperar todas terminarem, apenas a primeira, ou até que algumas completem:
import asyncio
async def tarefa_rapida():
await asyncio.sleep(1)
return "Rápida"
async def tarefa_lenta():
await asyncio.sleep(3)
return "Lenta"
async def main():
tasks = [
asyncio.create_task(tarefa_rapida()),
asyncio.create_task(tarefa_lenta())
]
# Retorna assim que a primeira task completar
done, pending = await asyncio.wait(
tasks,
return_when=asyncio.FIRST_COMPLETED
)
print(f"Primeira task concluída: {done.pop().result()}")
# Cancela tasks pendentes
for task in pending:
task.cancel()
asyncio.run(main())asyncio.as_completed()
Esta função retorna um iterador que produz resultados à medida que as tarefas vão sendo concluídas, permitindo processar resultados assim que ficam disponíveis:
import asyncio
async def buscar_dado(id, tempo):
await asyncio.sleep(tempo)
return f"Dado {id}"
async def main():
tarefas = [
buscar_dado(1, 3),
buscar_dado(2, 1),
buscar_dado(3, 2)
]
# Processa resultados conforme ficam prontos
for corrotina in asyncio.as_completed(tarefas):
resultado = await corrotina
print(f"Recebido: {resultado}")
asyncio.run(main())
# Saída:
# Recebido: Dado 2 (após 1s)
# Recebido: Dado 3 (após 2s)
# Recebido: Dado 1 (após 3s)Exemplo Real: Fazendo Requisições HTTP Assíncronas
Um dos usos mais comuns do asyncio é fazer múltiplas requisições HTTP de forma eficiente. Para isso, precisamos de uma biblioteca que suporte operações assíncronas. A mais popular é a aiohttp.
Primeiro, instale a biblioteca:
pip install aiohttpAgora vamos criar um exemplo que busca dados de múltiplas APIs simultaneamente:
import asyncio
import aiohttp
import time
async def buscar_post(session, post_id):
url = f"https://jsonplaceholder.typicode.com/posts/{post_id}"
async with session.get(url) as response:
data = await response.json()
return data
async def main():
inicio = time.time()
# Cria uma sessão compartilhada
async with aiohttp.ClientSession() as session:
# Cria tasks para buscar 10 posts
tasks = []
for i in range(1, 11):
task = asyncio.create_task(buscar_post(session, i))
tasks.append(task)
# Aguarda todos os resultados
posts = await asyncio.gather(*tasks)
# Exibe títulos
for post in posts:
print(f"Post {post['id']}: {post['title']}")
fim = time.time()
print(f"\nTempo total: {fim - inicio:.2f} segundos")
asyncio.run(main())Este código faz 10 requisições HTTP simultaneamente, algo que seria muito mais lento se feito de forma síncrona com a biblioteca requests tradicional.
Tratamento de Erros em Código Assíncrono
O tratamento de erros em código assíncrono funciona de forma semelhante ao código síncrono, usando try e except. No entanto, há algumas considerações especiais:
import asyncio
async def operacao_que_falha():
await asyncio.sleep(1)
raise ValueError("Algo deu errado!")
async def operacao_segura():
try:
resultado = await operacao_que_falha()
return resultado
except ValueError as e:
print(f"Erro capturado: {e}")
return None
async def main():
resultado = await operacao_segura()
print(f"Resultado: {resultado}")
asyncio.run(main())Quando você usa asyncio.gather(), pode especificar o parâmetro return_exceptions=True para coletar exceções junto com os resultados, em vez de interromper a execução:
import asyncio
async def tarefa_normal():
await asyncio.sleep(1)
return "Sucesso"
async def tarefa_com_erro():
await asyncio.sleep(0.5)
raise ValueError("Erro na tarefa")
async def main():
resultados = await asyncio.gather(
tarefa_normal(),
tarefa_com_erro(),
tarefa_normal(),
return_exceptions=True
)
for i, resultado in enumerate(resultados):
if isinstance(resultado, Exception):
print(f"Tarefa {i}: Erro - {resultado}")
else:
print(f"Tarefa {i}: {resultado}")
asyncio.run(main())Timeouts e Cancelamento de Tarefas
É fundamental ter controle sobre quanto tempo suas operações assíncronas podem levar. O asyncio fornece mecanismos robustos para definir timeouts e cancelar tarefas:
Usando asyncio.wait_for()
import asyncio
async def operacao_demorada():
await asyncio.sleep(5)
return "Concluído"
async def main():
try:
# Define timeout de 3 segundos
resultado = await asyncio.wait_for(
operacao_demorada(),
timeout=3.0
)
print(resultado)
except asyncio.TimeoutError:
print("Operação excedeu o tempo limite!")
asyncio.run(main())Cancelando Tarefas Manualmente
import asyncio
async def tarefa_longa():
try:
print("Iniciando tarefa...")
await asyncio.sleep(10)
print("Tarefa concluída")
except asyncio.CancelledError:
print("Tarefa foi cancelada!")
raise
async def main():
# Cria a task
task = asyncio.create_task(tarefa_longa())
# Aguarda 2 segundos
await asyncio.sleep(2)
# Cancela a task
task.cancel()
try:
await task
except asyncio.CancelledError:
print("Task cancelada com sucesso")
asyncio.run(main())Asyncio com Web Scraping
O asyncio é extremamente útil para web scraping, permitindo extrair dados de múltiplos sites simultaneamente. Vamos criar um exemplo prático:
import asyncio
import aiohttp
from bs4 import BeautifulSoup
async def extrair_titulo(session, url):
try:
async with session.get(url, timeout=10) as response:
html = await response.text()
soup = BeautifulSoup(html, 'html.parser')
titulo = soup.find('title')
return {
'url': url,
'titulo': titulo.text if titulo else 'Sem título'
}
except Exception as e:
return {'url': url, 'erro': str(e)}
async def main():
urls = [
'https://www.python.org',
'https://docs.python.org',
'https://pypi.org'
]
async with aiohttp.ClientSession() as session:
tasks = [extrair_titulo(session, url) for url in urls]
resultados = await asyncio.gather(*tasks)
for resultado in resultados:
if 'erro' in resultado:
print(f"{resultado['url']}: Erro - {resultado['erro']}")
else:
print(f"{resultado['url']}: {resultado['titulo']}")
asyncio.run(main())Boas Práticas ao Usar Asyncio
Para aproveitar ao máximo o asyncio e evitar problemas comuns, siga estas recomendações:
Use asyncio.run() para iniciar seu programa: Esta é a forma recomendada de executar código assíncrono a partir do Python 3.7. Ela cuida de criar e fechar o event loop automaticamente.
Evite misturar código bloqueante com assíncrono: Nunca use time.sleep() em código assíncrono. Use sempre await asyncio.sleep(). Operações bloqueantes como leitura de arquivos síncronos podem prejudicar o desempenho.
Gerencie seus recursos corretamente: Use async with para garantir que recursos como conexões de rede sejam fechados adequadamente, mesmo se ocorrerem erros.
Não se esqueça de await: Um erro comum é chamar uma corrotina sem usar await. Isso não executa a corrotina, apenas cria o objeto.
# Errado - corrotina não é executada
resultado = minha_corrotina()
# Correto - corrotina é executada
resultado = await minha_corrotina()Trate exceções adequadamente: Sempre implemente tratamento de erros robusto, especialmente ao trabalhar com operações de rede que podem falhar.
Use bibliotecas assíncronas: Para aproveitar os benefícios do asyncio, você precisa usar bibliotecas que suportam operações assíncronas, como aiohttp para HTTP, asyncpg para PostgreSQL, e motor para MongoDB.
Quando NÃO Usar Asyncio
Embora o asyncio seja poderoso, nem sempre é a melhor escolha. Aqui estão situações onde você deve considerar outras abordagens:
Tarefas CPU-bound: Se seu código faz cálculos pesados que realmente usam o processador (como processamento de imagens, cálculos matemáticos complexos ou machine learning), asyncio não ajudará. Nesses casos, use multiprocessing ou threading.
Bibliotecas não assíncronas: Se você precisa usar uma biblioteca que não suporta operações assíncronas e não há alternativa assíncrona disponível, forçar o uso de asyncio pode complicar desnecessariamente seu código.
Projetos simples: Para scripts pequenos e simples que não fazem muitas operações I/O, adicionar asyncio pode ser um exagero. A simplicidade do código síncrono pode ser mais valiosa.
Equipes sem experiência: O asyncio adiciona complexidade ao código. Se sua equipe não tem experiência com programação assíncrona, considere se os benefícios de desempenho justificam a curva de aprendizado.
Asyncio vs Threading vs Multiprocessing
É importante entender quando usar cada abordagem de concorrência no Python:
Asyncio: Ideal para operações I/O-bound com muitas tarefas simultâneas. Baixo overhead de memória, executa em um único thread. Perfeito para requisições de rede, operações de banco de dados e servidores web.
Threading: Bom para operações I/O-bound quando você não pode usar asyncio (bibliotecas não assíncronas). Tem mais overhead que asyncio mas ainda é eficiente para I/O. Limitado pelo GIL (Global Interpreter Lock) do Python.
Multiprocessing: Necessário para tarefas CPU-bound que realmente precisam de processamento paralelo. Contorna o GIL criando processos separados. Mais overhead de memória e comunicação entre processos mais complexa.
| Característica | Asyncio | Threading | Multiprocessing |
|---|---|---|---|
| Melhor para | I/O-bound | I/O-bound | CPU-bound |
| Overhead | Muito baixo | Médio | Alto |
| Complexidade | Média | Baixa | Alta |
| Tarefas simultâneas | Milhares | Dezenas | Poucos |
| Compartilhamento de memória | Fácil | Fácil | Difícil |
Perguntas Frequentes (FAQ)
1. O asyncio funciona em todas as versões do Python?
O asyncio está disponível desde o Python 3.4, mas a sintaxe moderna com async/await foi introduzida no Python 3.5. Recomenda-se usar Python 3.7 ou superior para melhor experiência.
2. Asyncio é o mesmo que multithreading?
Não. O asyncio executa tudo em um único thread, alternando entre tarefas quando há espera. Multithreading cria múltiplos threads que podem executar simultaneamente.
3. Posso usar asyncio com Flask ou Django?
O Flask tradicional não é assíncrono, mas você pode usar Quart como alternativa. O Django 3.0+ tem suporte parcial a asyncio. Para projetos novos focados em performance, considere FastAPI.
4. Como posso converter código síncrono para assíncrono?
Adicione async antes das definições de função, substitua chamadas bloqueantes por versões assíncronas (time.sleep por asyncio.sleep) e adicione await antes de chamadas assíncronas.
5. O asyncio melhora o desempenho de qualquer código?
Não. O asyncio é benéfico apenas para operações I/O-bound. Para tarefas CPU-bound, use multiprocessing. Para código simples sem I/O, pode até adicionar complexidade desnecessária.
6. Preciso instalar o asyncio separadamente?
Não. O asyncio é uma biblioteca padrão do Python desde a versão 3.4 e já vem incluído na instalação básica do Python.
7. Como depurar código assíncrono?
Use o modo debug do asyncio (asyncio.run(main(), debug=True)), adicione logging estratégico e use ferramentas como o debugger do VS Code com suporte a código assíncrono.
8. Posso misturar código síncrono e assíncrono?
Sim, mas com cuidado. Você pode chamar código síncrono de dentro de código assíncrono, mas isso pode bloquear o event loop. Use asyncio.to_thread() para executar código bloqueante em thread separada.
9. Quantas tarefas posso executar simultaneamente com asyncio?
Milhares! O asyncio tem overhead muito baixo. É comum ver aplicações gerenciando 10.000+ conexões simultâneas, algo impraticável com threads tradicionais.
10. O asyncio funciona no Windows?
Sim, mas com algumas limitações. O Windows usa ProactorEventLoop por padrão. A maioria das funcionalidades funciona normalmente, mas alguns recursos avançados podem ter comportamento diferente.
11. Como lidar com operações de banco de dados com asyncio?
Use drivers de banco de dados assíncronos como asyncpg para PostgreSQL, motor para MongoDB ou aiomysql para MySQL. Drivers síncronos tradicionais bloquearão o event loop.
12. O asyncio consome muita memória?
Na verdade, o asyncio é muito eficiente em termos de memória comparado a threads. Cada thread consome cerca de 8MB de memória, enquanto corrotinas consomem apenas alguns KB.
