asyncio.Semaphore: Limite Concorrência em Python

Criar milhares de corrotinas não significa que uma API, banco ou pool aceite milhares de operações simultâneas. asyncio.Semaphore mantém um contador de permissões e faz novas tarefas aguardarem quando o limite é atingido.

import asyncio

async def processar(item: int, limite: asyncio.Semaphore) -> int:
    async with limite:
        await asyncio.sleep(0.1)
        return item * 2

async def main() -> None:
    limite = asyncio.Semaphore(5)
    resultados = await asyncio.gather(
        *(processar(item, limite) for item in range(20))
    )
    print(resultados)

asyncio.run(main())

O bloco async with adquire e libera a permissão de forma segura. Escolha o valor com base no recurso real, como tamanho do pool e limite documentado do serviço, e meça latência, erros e backlog antes de aumentar.

Boas práticas

Semaphore controla concorrência instantânea, não requisições por segundo. Para uma cota temporal, combine uma política de rate limiting adequada. Coloque somente a operação limitada dentro do bloco para não segurar permissões durante trabalho local. Se o valor inicial não puder ser excedido por engano, avalie BoundedSemaphore.

Para organizar tarefas relacionadas, veja asyncio.TaskGroup. Em chamadas HTTP, aplique limites também ao cliente descrito em HTTPX em Python.

A documentação oficial das primitivas de sincronização asyncio, consultada em 22 de julho de 2026, detalha a API e seus limites.