Arquivo de Paralela ou Concorrente

Async e Await em Python

22/05/202622/04/2026 Por antonino

Async e await são palavras-chave para programação assíncrona em Python. Elas foram introduzidas no Python 3.5 para facilitar a concorrência. Primeiramente, async def define uma função assíncrona (corrotina). Dentro dela, await pausa a execução sem bloquear a thread. Por exemplo, await asyncio.sleep(1) espera 1 segundo cooperativamente. Além disso, você só pode usar await dentro de funções async. Assim, o código se torna mais claro e previsível. Consequentemente, a legibilidade melhora drasticamente. Quando utilizar async/await? Em operações de I/O com alta concorrência. Por exemplo, requisições HTTP, acesso a banco de dados ou WebSockets. Por outro lado, async/await não é para tarefas com CPU intensiva. Portanto, identifique primeiro o tipo do seu gargalo. Vamos explorar sintaxe, padrões e boas práticas. Três subtítulos guiarão você pelo coração da programação assíncrona. Finalmente, você dominará async e await em Python.

Sintaxe básica: async def, await e asyncio.run

Uma corrotina é definida com async def nome(). Para chamar uma corrotina, você precisa usar await. O ponto de entrada principal usa asyncio.run(corrotina()). Então, isso cria um event loop e executa a corrotina até completar. Quando usar funções async? Sempre que houver espera por I/O. Por exemplo, leitura de arquivos ou chamadas de rede. Exemplo básico mostrando a sintaxe fundamental:

import asyncio
import time

# Definindo uma corrotina (função assíncrona)
async def saudacao_assincrona(nome, segundos):
    """Função que espera e retorna uma saudação."""
    print(f"  Iniciando saudação para {nome}")
    await asyncio.sleep(segundos)  # Pausa cooperativa
    print(f"  Saudação para {nome} concluída após {segundos}s")
    return f"Olá, {nome}!"

async def tarefa_rapida():
    """Corrotina que não espera muito."""
    print("  Tarefa rápida: começando")
    await asyncio.sleep(0.5)
    print("  Tarefa rápida: terminou")
    return "Rápido"

async def tarefa_lenta():
    """Corrotina que demora mais."""
    print("  Tarefa lenta: começando")
    await asyncio.sleep(2)
    print("  Tarefa lenta: terminou")
    return "Lento"

async def main():
    print("=== Execução sequencial (uma await após outra) ===")
    inicio = time.time()
    
    resultado1 = await saudacao_assincrona("Ana", 1)
    resultado2 = await saudacao_assincrona("Bob", 1)
    
    print(f"Resultados sequenciais: {resultado1}, {resultado2}")
    print(f"Tempo sequencial: {time.time() - inicio:.1f}s")
    
    print("\n=== Execução concorrente (múltiplos awaits em tasks) ===")
    inicio = time.time()
    
    # Criando tasks para executar concorrentemente
    task1 = asyncio.create_task(saudacao_assincrona("Carlos", 1))
    task2 = asyncio.create_task(saudacao_assincrona("Diana", 1))
    
    # Aguardando ambas
    resultado3 = await task1
    resultado4 = await task2
    
    print(f"Resultados concorrentes: {resultado3}, {resultado4}")
    print(f"Tempo concorrente: {time.time() - inicio:.1f}s")
    
    print("\n=== gather: execute múltiplas corrotinas em paralelo ===")
    inicio = time.time()
    
    resultados = await asyncio.gather(
        tarefa_rapida(),
        tarefa_lenta(),
        saudacao_assincrona("Eva", 1)
    )
    
    print(f"Resultados do gather: {resultados}")
    print(f"Tempo do gather: {time.time() - inicio:.1f}s")

if __name__ == "__main__":
    asyncio.run(main())

import asyncio

import time

# Definindo uma corrotina (função assíncrona)

async def saudacao_assincrona(nome, segundos):

"""Função que espera e retorna uma saudação."""

print(f" Iniciando saudação para {nome}")

await asyncio.sleep(segundos) # Pausa cooperativa

print(f" Saudação para {nome} concluída após {segundos}s")

return f"Olá, {nome}!"

async def tarefa_rapida():

"""Corrotina que não espera muito."""

print(" Tarefa rápida: começando")

await asyncio.sleep(0.5)

print(" Tarefa rápida: terminou")

return "Rápido"

async def tarefa_lenta():

"""Corrotina que demora mais."""

print(" Tarefa lenta: começando")

await asyncio.sleep(2)

print(" Tarefa lenta: terminou")

return "Lento"

async def main():

print("=== Execução sequencial (uma await após outra) ===")

inicio = time.time()

resultado1 = await saudacao_assincrona("Ana", 1)

resultado2 = await saudacao_assincrona("Bob", 1)

print(f"Resultados sequenciais: {resultado1}, {resultado2}")

print(f"Tempo sequencial: {time.time() - inicio:.1f}s")

print("\n=== Execução concorrente (múltiplos awaits em tasks) ===")

inicio = time.time()

# Criando tasks para executar concorrentemente

task1 = asyncio.create_task(saudacao_assincrona("Carlos", 1))

task2 = asyncio.create_task(saudacao_assincrona("Diana", 1))

# Aguardando ambas

resultado3 = await task1

resultado4 = await task2

print(f"Resultados concorrentes: {resultado3}, {resultado4}")

print(f"Tempo concorrente: {time.time() - inicio:.1f}s")

print("\n=== gather: execute múltiplas corrotinas em paralelo ===")

inicio = time.time()

resultados = await asyncio.gather(

tarefa_rapida(),

tarefa_lenta(),

saudacao_assincrona("Eva", 1)

)

print(f"Resultados do gather: {resultados}")

print(f"Tempo do gather: {time.time() - inicio:.1f}s")

if __name__ == "__main__":

asyncio.run(main())

Use create_task() para agendar corrotinas concorrentes. Use gather() para esperar várias corrotinas simultaneamente. Nunca esqueça de await em chamadas de corrotinas. Assim, você evita erros comuns de iniciantes.

Padrões avançados: timeout, wait e as_completed

Async/await oferece ferramentas para controlar a execução. asyncio.wait_for() adiciona timeout a uma corrotina. asyncio.wait() espera múltiplas tarefas com condições. asyncio.as_completed() processa resultados conforme terminam. Quando usar esses padrões? Em sistemas que exigem limites de tempo. Também em pipelines onde você quer respostas parciais. Além disso, timeouts previnem travamentos eternos. Exemplo prático com timeout e processamento parcial:

import asyncio
import random

async def tarefa_instavel(nome, tempo_espera, chance_erro=0):
    """Tarefa que pode demorar ou falhar."""
    print(f"  {nome}: iniciando (vai demorar {tempo_espera}s)")
    await asyncio.sleep(tempo_espera)
    
    if chance_erro > 0 and random.random() < chance_erro:
        raise Exception(f"{nome}: Falhou após {tempo_espera}s")
    
    print(f"  {nome}: concluída")
    return f"Resultado de {nome}"

async def main():
    print("=== Exemplo com timeout ===")
    try:
        # Timeout de 2 segundos para uma tarefa que demora 3
        resultado = await asyncio.wait_for(
            tarefa_instavel("Tarefa-Lenta", 3), 
            timeout=2
        )
        print(f"Resultado: {resultado}")
    except asyncio.TimeoutError:
        print("Timeout! Tarefa levou mais que 2 segundos")
    
    print("\n=== Exemplo com wait (FIRST_COMPLETED) ===")
    tarefas = [
        tarefa_instavel("A", 3),
        tarefa_instavel("B", 1),
        tarefa_instavel("C", 2),
    ]
    
    # Aguarda a primeira tarefa completar
    concluidas, pendentes = await asyncio.wait(
        tarefas, 
        return_when=asyncio.FIRST_COMPLETED
    )
    
    for tarefa in concluidas:
        resultado = tarefa.result()
        print(f"Primeira tarefa concluída: {resultado}")
    
    # Cancela as tarefas pendentes
    for tarefa in pendentes:
        tarefa.cancel()
        print("Tarefas pendentes foram canceladas")
    
    print("\n=== Exemplo com as_completed (processamento parcial) ===")
    tarefas = [
        tarefa_instavel("X", 2),
        tarefa_instavel("Y", 1),
        tarefa_instavel("Z", 3),
    ]
    
    # Processa resultados conforme terminam
    inicio = time.time()
    for corrotina in asyncio.as_completed(tarefas):
        try:
            resultado = await corrotina
            print(f"Resultado recebido em {time.time() - inicio:.1f}s: {resultado}")
        except Exception as e:
            print(f"Erro: {e}")
    
    print("\n=== Exemplo com wait_for e shields (proteção) ===")
    # asyncio.shield protege uma tarefa de ser cancelada
    tarefa_protegida = asyncio.create_task(tarefa_instavel("Protegida", 2))
    
    try:
        resultado = await asyncio.wait_for(asyncio.shield(tarefa_protegida), timeout=1)
    except asyncio.TimeoutError:
        print("Timeout atingido, mas a tarefa protegida continua")
        # A tarefa ainda está rodando em background
        resultado_final = await tarefa_protegida
        print(f"Resultado final da tarefa protegida: {resultado_final}")

if __name__ == "__main__":
    import time
    asyncio.run(main())

import asyncio

import random

async def tarefa_instavel(nome, tempo_espera, chance_erro=0):

"""Tarefa que pode demorar ou falhar."""

print(f" {nome}: iniciando (vai demorar {tempo_espera}s)")

await asyncio.sleep(tempo_espera)

if chance_erro > 0 and random.random() < chance_erro:

raise Exception(f"{nome}: Falhou após {tempo_espera}s")

print(f" {nome}: concluída")

return f"Resultado de {nome}"

async def main():

print("=== Exemplo com timeout ===")

try:

# Timeout de 2 segundos para uma tarefa que demora 3

resultado = await asyncio.wait_for(

tarefa_instavel("Tarefa-Lenta", 3),

timeout=2

)

print(f"Resultado: {resultado}")

except asyncio.TimeoutError:

print("Timeout! Tarefa levou mais que 2 segundos")

print("\n=== Exemplo com wait (FIRST_COMPLETED) ===")

tarefas = [

tarefa_instavel("A", 3),

tarefa_instavel("B", 1),

tarefa_instavel("C", 2),

]

# Aguarda a primeira tarefa completar

concluidas, pendentes = await asyncio.wait(

tarefas,

return_when=asyncio.FIRST_COMPLETED

)

for tarefa in concluidas:

resultado = tarefa.result()

print(f"Primeira tarefa concluída: {resultado}")

# Cancela as tarefas pendentes

for tarefa in pendentes:

tarefa.cancel()

print("Tarefas pendentes foram canceladas")

print("\n=== Exemplo com as_completed (processamento parcial) ===")

tarefas = [

tarefa_instavel("X", 2),

tarefa_instavel("Y", 1),

tarefa_instavel("Z", 3),

]

# Processa resultados conforme terminam

inicio = time.time()

for corrotina in asyncio.as_completed(tarefas):

try:

resultado = await corrotina

print(f"Resultado recebido em {time.time() - inicio:.1f}s: {resultado}")

except Exception as e:

print(f"Erro: {e}")

print("\n=== Exemplo com wait_for e shields (proteção) ===")

# asyncio.shield protege uma tarefa de ser cancelada

tarefa_protegida = asyncio.create_task(tarefa_instavel("Protegida", 2))

try:

resultado = await asyncio.wait_for(asyncio.shield(tarefa_protegida), timeout=1)

except asyncio.TimeoutError:

print("Timeout atingido, mas a tarefa protegida continua")

# A tarefa ainda está rodando em background

resultado_final = await tarefa_protegida

print(f"Resultado final da tarefa protegida: {resultado_final}")

if __name__ == "__main__":

import time

asyncio.run(main())

Timeouts previnem travamentos em operações lentas. as_completed é ótimo para mostrar progresso gradual. shield protege tarefas importantes de cancelamento. Portanto, use essas ferramentas para sistemas robustos.

Async/await vs. threads vs. callbacks tradicionais

Async/await é muito mais legível que callbacks aninhados. Ele também usa menos memória que threads para muitas conexões. No entanto, threads são melhores para código síncrono legado. Quando escolher async/await? Em projetos novos com muito I/O. Também quando você precisa de milhares de conexões simultâneas. Além disso, async/await facilita o tratamento de erros. A fórmula de legibilidade mostra vantagem clara: \(L = \frac{N_{\text{linhas}}}{N_{\text{aninhamento}}}\) Async/await reduz aninhamento comparado a callbacks. Exemplo comparando callbacks vs. async/await:

import asyncio
import time

# Versão com callbacks (estilo antigo, difícil)
def baixar_com_callback(url, callback):
    """Simula download com callback."""
    def executar():
        time.sleep(1)  # Simula I/O bloqueante (ruim)
        callback(f"Dados de {url}")
    import threading
    threading.Thread(target=executar).start()

def callback_antigo(resultado):
    print(f"Callback recebeu: {resultado}")
    # Callback hell começaria aqui se tivesse mais chamadas

# Versão com async/await (moderno e legível)
async def baixar_assincrono(url):
    """Simula download com async/await."""
    await asyncio.sleep(1)  # I/O não bloqueante
    return f"Dados de {url}"

async def processar_em_paralelo(urls):
    """Processa múltiplos downloads de forma legível."""
    tarefas = [baixar_assincrono(url) for url in urls]
    resultados = await asyncio.gather(*tarefas)
    
    for resultado in resultados:
        print(f"Processando: {resultado}")
    
    # Fácil adicionar transformações
    dados_upper = [r.upper() for r in resultados]
    return dados_upper

async def main():
    print("=== Callbacks (difícil de ler e manter) ===")
    baixar_com_callback("site1.com", callback_antigo)
    baixar_com_callback("site2.com", callback_antigo)
    time.sleep(2)  # Espera feia
    
    print("\n=== Async/Await (claro e legível) ===")
    urls = ["site1.com", "site2.com", "site3.com"]
    resultados = await processar_em_paralelo(urls)
    print(f"Resultados finais: {resultados}")
    
    print("\n=== Async/Await com tratamento de erros elegante ===")
    async def tarefa_segura():
        try:
            await asyncio.sleep(0.5)
            return "Sucesso"
        except Exception as e:
            return f"Erro: {e}"
    
    resultados = await asyncio.gather(
        tarefa_segura(),
        tarefa_segura(),
        return_exceptions=True  # Não levanta exceção imediatamente
    )
    print(f"Resultados com exceptions capturadas: {resultados}")

if __name__ == "__main__":
    asyncio.run(main())

import asyncio

import time

# Versão com callbacks (estilo antigo, difícil)

def baixar_com_callback(url, callback):

"""Simula download com callback."""

def executar():

time.sleep(1) # Simula I/O bloqueante (ruim)

callback(f"Dados de {url}")

import threading

threading.Thread(target=executar).start()

def callback_antigo(resultado):

print(f"Callback recebeu: {resultado}")

# Callback hell começaria aqui se tivesse mais chamadas

# Versão com async/await (moderno e legível)

async def baixar_assincrono(url):

"""Simula download com async/await."""

await asyncio.sleep(1) # I/O não bloqueante

return f"Dados de {url}"

async def processar_em_paralelo(urls):

"""Processa múltiplos downloads de forma legível."""

tarefas = [baixar_assincrono(url) for url in urls]

resultados = await asyncio.gather(*tarefas)

for resultado in resultados:

print(f"Processando: {resultado}")

# Fácil adicionar transformações

dados_upper = [r.upper() for r in resultados]

return dados_upper

async def main():

print("=== Callbacks (difícil de ler e manter) ===")

baixar_com_callback("site1.com", callback_antigo)

baixar_com_callback("site2.com", callback_antigo)

time.sleep(2) # Espera feia

print("\n=== Async/Await (claro e legível) ===")

urls = ["site1.com", "site2.com", "site3.com"]

resultados = await processar_em_paralelo(urls)

print(f"Resultados finais: {resultados}")

print("\n=== Async/Await com tratamento de erros elegante ===")

async def tarefa_segura():

try:

await asyncio.sleep(0.5)

return "Sucesso"

except Exception as e:

return f"Erro: {e}"

resultados = await asyncio.gather(

tarefa_segura(),

return_exceptions=True # Não levanta exceção imediatamente

)

print(f"Resultados com exceptions capturadas: {resultados}")

if __name__ == "__main__":

asyncio.run(main())

Async/await torna o código assíncrono tão legível quanto síncrono. Use return_exceptions=True no gather para lidar com falhas. A regra de ouro: toda operação de I/O deve ser async. Comece com async/await em novas funções de rede. Gradualmente, converta código síncrono legado. Você nunca mais vai querer voltar para callbacks. Portanto, abrace async/await em seus próximos projetos.

Assíncrona em Python

22/05/202622/04/2026 Por antonino

Asyncio é uma biblioteca para escrever código concorrente com async/await. Ela usa um event loop para gerenciar múltiplas tarefas em uma única thread. Primeiramente, isso difere de threads e processos tradicionais. Por exemplo, uma tarefa pode esperar dados da rede sem bloquear as outras. Além disso, asyncio é extremamente leve e escala para milhares de conexões. Consequentemente, a memória consumida é muito menor que em threads. Assim, você consegue alta concorrência sem custo excessivo. Então, quando utilizar asyncio? Em aplicações com muito I/O. Por exemplo, servidores web, chats, proxies ou crawlers. Da mesma forma, é ótimo para APIs que chamam outras APIs externas. No entanto, asyncio não é indicado para tarefas com CPU intensiva. Portanto, identifique primeiro o tipo do seu gargalo. Vamos explorar conceitos, padrões e boas práticas. Três subtítulos guiarão você pelo mundo assíncrono em Python. Finalmente, você construirá aplicações eficientes e escaláveis.

Conceitos fundamentais: corrotinas, tasks e event loop

Corrotinas são funções definidas com async def. Elas podem pausar sua execução com await sem bloquear a thread. Tasks são corrotinas agendadas no event loop para execução. O event loop gerencia todas as tasks e retoma cada uma no momento certo. Assim, você consegue concorrência cooperativa muito eficiente. Por outro lado, programadores iniciantes podem achar o conceito complexo. Então, quando usar corrotinas? Sempre que você tiver operações de I/O. Por exemplo, chamadas de rede, acesso a banco ou leitura de arquivos. Exemplo básico mostrando corrotinas e event loop:

import asyncio
import time

async def dizer_ola(nome, atraso):
    """Corrotina que espera e depois imprime."""
    print(f"Iniciando saudação para {nome}")
    await asyncio.sleep(atraso)  # Simula I/O, não bloqueia
    print(f"Olá, {nome}! (após {atraso}s)")
    return f"Resultado de {nome}"

async def main():
    print(f"Início: {time.strftime('%H:%M:%S')}")
    
    # Executando corrotinas sequencialmente
    resultado1 = await dizer_ola("Ana", 2)
    resultado2 = await dizer_ola("Bob", 1)
    print(f"Sequencial: {resultado1}, {resultado2}")
    
    print(f"\nExecutando concorrentemente:")
    inicio_conc = time.time()
    
    # Criando tasks para execução concorrente
    task1 = asyncio.create_task(dizer_ola("Carlos", 2))
    task2 = asyncio.create_task(dizer_ola("Diana", 1))
    task3 = asyncio.create_task(dizer_ola("Eva", 1.5))
    
    # Aguardando todas completarem
    resultados = await asyncio.gather(task1, task2, task3)
    tempo_conc = time.time() - inicio_conc
    print(f"Resultados concorrentes: {resultados}")
    print(f"Tempo concorrente: {tempo_conc:.1f}s (equivalente à tarefa mais longa)")
    
    # Usando as_completed para processar conforme terminam
    print("\nProcessando conforme terminam:")
    tarefas = [
        dizer_ola("Fábio", 2),
        dizer_ola("Gina", 1),
        dizer_ola("Hugo", 1.8)
    ]
    
    for corrotina in asyncio.as_completed(tarefas):
        resultado = await corrotina
        print(f"Resultado recebido: {resultado}")

if __name__ == "__main__":
    asyncio.run(main())

import asyncio

import time

async def dizer_ola(nome, atraso):

"""Corrotina que espera e depois imprime."""

print(f"Iniciando saudação para {nome}")

await asyncio.sleep(atraso) # Simula I/O, não bloqueia

print(f"Olá, {nome}! (após {atraso}s)")

return f"Resultado de {nome}"

async def main():

print(f"Início: {time.strftime('%H:%M:%S')}")

# Executando corrotinas sequencialmente

resultado1 = await dizer_ola("Ana", 2)

resultado2 = await dizer_ola("Bob", 1)

print(f"Sequencial: {resultado1}, {resultado2}")

print(f"\nExecutando concorrentemente:")

inicio_conc = time.time()

# Criando tasks para execução concorrente

task1 = asyncio.create_task(dizer_ola("Carlos", 2))

task2 = asyncio.create_task(dizer_ola("Diana", 1))

task3 = asyncio.create_task(dizer_ola("Eva", 1.5))

# Aguardando todas completarem

resultados = await asyncio.gather(task1, task2, task3)

tempo_conc = time.time() - inicio_conc

print(f"Resultados concorrentes: {resultados}")

print(f"Tempo concorrente: {tempo_conc:.1f}s (equivalente à tarefa mais longa)")

# Usando as_completed para processar conforme terminam

print("\nProcessando conforme terminam:")

tarefas = [

dizer_ola("Fábio", 2),

dizer_ola("Gina", 1),

dizer_ola("Hugo", 1.8)

]

for corrotina in asyncio.as_completed(tarefas):

resultado = await corrotina

print(f"Resultado recebido: {resultado}")

if __name__ == "__main__":

asyncio.run(main())

Tasks concorrentes executam em paralelo lógico, não físico. O tempo total é o da tarefa mais longa, não a soma de todas. Portanto, esse é o grande poder do asyncio para I/O.

Padrões práticos: produtor-consumidor e timeouts

Asyncio oferece filas (asyncio.Queue) para comunicação entre tasks. Assim, você implementa facilmente o padrão produtor-consumidor assíncrono. Timeouts são essenciais para evitar bloqueios eternos. Use asyncio.wait_for() para limitar o tempo de espera. Quando usar filas? Em pipelines de processamento de dados. Por exemplo, crawlers web com múltiplas etapas. Exemplo de produtor-consumidor com asyncio:

import asyncio
import random

async def produtor(queue, id_produtor, n_itens):
    """Produz itens e coloca na fila."""
    for i in range(n_itens):
        item = f"P{id_produtor}-Item{i}"
        await queue.put(item)
        print(f"[Produtor {id_produtor}] Produziu: {item}")
        await asyncio.sleep(random.uniform(0.1, 0.3))
    print(f"[Produtor {id_produtor}] Finalizou")

async def consumidor(queue, id_consumidor, nome):
    """Consome itens da fila."""
    while True:
        try:
            # Timeout para evitar espera infinita
            item = await asyncio.wait_for(queue.get(), timeout=1.0)
            print(f"[Consumidor {id_consumidor} ({nome})] Consumiu: {item}")
            # Simula processamento do item
            await asyncio.sleep(random.uniform(0.2, 0.5))
            queue.task_done()
        except asyncio.TimeoutError:
            print(f"[Consumidor {id_consumidor}] Timeout, verificando se terminou...")
            if queue.empty() and all(produtor_task.done() for produtor_task in produtor_tasks):
                break
            continue

async def main():
    queue = asyncio.Queue(maxsize=5)  # Fila com tamanho máximo
    
    global produtor_tasks
    # Produtores
    produtor_tasks = []
    for i in range(2):
        p = asyncio.create_task(produtor(queue, i, 5))
        produtor_tasks.append(p)
    
    # Consumidores
    consumidor_tasks = []
    for i in range(3):
        c = asyncio.create_task(consumidor(queue, i, f"Cons-{i}"))
        consumidor_tasks.append(c)
    
    # Aguardar produtores
    await asyncio.gather(*produtor_tasks)
    print("\nProdutores finalizaram. Aguardando consumidores...")
    
    # Aguardar consumidores
    await asyncio.gather(*consumidor_tasks)
    print("Todos os consumidores finalizaram")
    
    # Exemplo de timeout em operação
    print("\n=== Exemplo de Timeout ===")
    try:
        resultado = await asyncio.wait_for(asyncio.sleep(3), timeout=1)
    except asyncio.TimeoutError:
        print("Timeout! Operação levou mais de 1 segundo")

if __name__ == "__main__":
    asyncio.run(main())

import asyncio

import random

async def produtor(queue, id_produtor, n_itens):

"""Produz itens e coloca na fila."""

for i in range(n_itens):

item = f"P{id_produtor}-Item{i}"

await queue.put(item)

print(f"[Produtor {id_produtor}] Produziu: {item}")

await asyncio.sleep(random.uniform(0.1, 0.3))

print(f"[Produtor {id_produtor}] Finalizou")

async def consumidor(queue, id_consumidor, nome):

"""Consome itens da fila."""

while True:

try:

# Timeout para evitar espera infinita

item = await asyncio.wait_for(queue.get(), timeout=1.0)

print(f"[Consumidor {id_consumidor} ({nome})] Consumiu: {item}")

# Simula processamento do item

await asyncio.sleep(random.uniform(0.2, 0.5))

queue.task_done()

except asyncio.TimeoutError:

print(f"[Consumidor {id_consumidor}] Timeout, verificando se terminou...")

if queue.empty() and all(produtor_task.done() for produtor_task in produtor_tasks):

break

continue

async def main():

queue = asyncio.Queue(maxsize=5) # Fila com tamanho máximo

global produtor_tasks

# Produtores

produtor_tasks = []

for i in range(2):

p = asyncio.create_task(produtor(queue, i, 5))

produtor_tasks.append(p)

# Consumidores

consumidor_tasks = []

for i in range(3):

c = asyncio.create_task(consumidor(queue, i, f"Cons-{i}"))

consumidor_tasks.append(c)

# Aguardar produtores

await asyncio.gather(*produtor_tasks)

print("\nProdutores finalizaram. Aguardando consumidores...")

# Aguardar consumidores

await asyncio.gather(*consumidor_tasks)

print("Todos os consumidores finalizaram")

# Exemplo de timeout em operação

print("\n=== Exemplo de Timeout ===")

try:

resultado = await asyncio.wait_for(asyncio.sleep(3), timeout=1)

except asyncio.TimeoutError:

print("Timeout! Operação levou mais de 1 segundo")

if __name__ == "__main__":

asyncio.run(main())

Filas assíncronas são thread-safe e perfeitas para produtor-consumidor. Timeouts previnem que tarefas travem o sistema indefinidamente. Assim, esses padrões são essenciais para sistemas robustos.

Asyncio vs. threading vs. multiprocessing

Asyncio é leve e escala para milhares de conexões simultâneas. Threading tem overhead maior e sofre com GIL para CPU. Multiprocessing é mais pesado, mas permite paralelismo verdadeiro. Quando escolher asyncio? Em servidores de rede com alta concorrência. Também em clientes que chamam muitas APIs externas. Por outro lado, para CPU intensivo, prefira multiprocessing. A fórmula de eficiência para I/O-bound é clara: \(E_{\text{asyncio}} \approx \frac{M_{\text{thread}}}{M_{\text{asyncio}}} \approx 10\) Portanto, asyncio pode usar 10x menos memória que threads. Exemplo comparativo de desempenho para I/O:

import asyncio
import threading
import time
import aiohttp
import requests

# Asyncio version
async def baixar_asyncio(urls):
    async with aiohttp.ClientSession() as session:
        tarefas = []
        for url in urls:
            tarefas.append(session.get(url))
        respostas = await asyncio.gather(*tarefas)
        return [await r.text() for r in respostas]

# Threading version
def baixar_thread(urls):
    def fetch(url, idx, resultados):
        resultados[idx] = requests.get(url).text
    
    resultados = [None] * len(urls)
    threads = []
    for i, url in enumerate(urls):
        t = threading.Thread(target=fetch, args=(url, i, resultados))
        threads.append(t)
        t.start()
    
    for t in threads:
        t.join()
    return resultados

# Simulando muitas requisições
async def main():
    urls = ['https://httpbin.org/delay/0.1'] * 20
    
    print("Asyncio (muitas conexões leves):")
    inicio = time.time()
    await baixar_asyncio(urls)
    print(f"Tempo: {time.time() - inicio:.2f}s")
    
    print("\nThreading (mesmas conexões):")
    inicio = time.time()
    baixar_thread(urls)
    print(f"Tempo: {time.time() - inicio:.2f}s")
    
    print("\n=== Servidor Echo Simples ===")
    # Servidor echo (apenas para demonstração)

if __name__ == "__main__":
    asyncio.run(main())

import asyncio

import threading

import time

import aiohttp

import requests

# Asyncio version

async def baixar_asyncio(urls):

async with aiohttp.ClientSession() as session:

tarefas = []

for url in urls:

tarefas.append(session.get(url))

respostas = await asyncio.gather(*tarefas)

return [await r.text() for r in respostas]

# Threading version

def baixar_thread(urls):

def fetch(url, idx, resultados):

resultados[idx] = requests.get(url).text

resultados = [None] * len(urls)

threads = []

for i, url in enumerate(urls):

t = threading.Thread(target=fetch, args=(url, i, resultados))

threads.append(t)

t.start()

for t in threads:

t.join()

return resultados

# Simulando muitas requisições

async def main():

urls = ['https://httpbin.org/delay/0.1'] * 20

print("Asyncio (muitas conexões leves):")

inicio = time.time()

await baixar_asyncio(urls)

print(f"Tempo: {time.time() - inicio:.2f}s")

print("\nThreading (mesmas conexões):")

inicio = time.time()

baixar_thread(urls)

print(f"Tempo: {time.time() - inicio:.2f}s")

print("\n=== Servidor Echo Simples ===")

# Servidor echo (apenas para demonstração)

if __name__ == "__main__":

asyncio.run(main())

Asyncio é a escolha moderna para aplicações I/O intensivas. Ele combina performance com simplicidade de código. Use asyncio.run() como ponto de entrada principal. Nunca misture código síncrono bloqueante dentro de corrotinas. Para tarefas CPU-bound, use asyncio.to_thread() para delegar. Portanto, domine asyncio e suas aplicações serão extremamente eficientes. Comece com exemplos pequenos e escale para sistemas reais. Finalmente, o futuro da concorrência em Python é assíncrono.