# API de CPF: como fazer load balancing entre múltiplas chaves de API

> Aprenda a distribuir consultas de CPF entre múltiplas chaves de API. Aumente a cota efetiva e adicione redundância à integração.

**Publicado:** 24/08/2026
**Autor:** Redação CPFHub.io
**URL:** https://cpfhub.io/blog/api-cpf-load-balancing-multiplas-chaves-api

---


Quando o volume de consultas de CPF ultrapassa a cota de uma única chave de API, ou quando a operação exige redundância para garantir disponibilidade, a solução é distribuir as requisições entre múltiplas chaves. Essa estratégia — conhecida como load balancing no nível da aplicação — maximiza a cota efetiva, adiciona uma camada extra de resiliência e permite segregar consumo por ambiente ou departamento sem depender de um único ponto de falha.

---

## Quando usar múltiplas chaves

### Cenários típicos

- **Volume alto distribuído entre departamentos:** diferentes áreas da empresa têm suas próprias cotas e orçamentos.
- **Redundância:** se uma chave for comprometida ou atingir o limite, outra assume automaticamente.
- **Ambientes separados:** chaves distintas para desenvolvimento, staging e produção.
- **Escala além do plano individual:** combinar múltiplas chaves Pro para atingir volume corporativo enquanto negocia um plano Corporativo.

### Considerações importantes

Antes de implementar, verifique os termos de uso do provedor. A [**CPFHub.io**](https://www.cpfhub.io/)

---

## Estratégia 1 -- Round-robin

A estratégia mais simples: distribui as requisições de forma circular entre as chaves disponíveis.

```python
import requests
import logging
from itertools import cycle
from threading import Lock
from typing import Optional, Dict, List

logger = logging.getLogger(__name__)

class RoundRobinBalancer:
 """Distribui consultas de CPF em round-robin entre múltiplas chaves."""

 BASE_URL = "https://api.cpfhub.io/cpf"
 TIMEOUT = 30

 def __init__(self, api_keys: List[str]):
 if not api_keys:
 raise ValueError("Pelo menos uma chave de API e necessaria")
 self._keys = api_keys
 self._cycle = cycle(range(len(api_keys)))
 self._lock = Lock()

 def _proxima_chave(self) -> str:
 with self._lock:
 idx = next(self._cycle)
 return self._keys[idx]

 def consultar(self, cpf: str) -> Optional[Dict]:
 chave = self._proxima_chave()
 cpf_limpo = cpf.replace(".", "").replace("-", "")

 try:
 response = requests.get(
 f"{self.BASE_URL}/{cpf_limpo}",
 headers={
 "x-api-key": chave,
 "Accept": "application/json"
 },
 timeout=self.TIMEOUT
 )
 response.raise_for_status()
 return response.json()

 except requests.exceptions.Timeout:
 logger.error("Timeout na consulta de CPF")
 return None
 except requests.exceptions.RequestException as e:
 logger.error(f"Erro: {e}")
 return None

# Uso
balancer = RoundRobinBalancer(api_keys=[
 "chave_pro_1",
 "chave_pro_2",
 "chave_pro_3"
])

resultado = balancer.consultar("12345678900")
```

### Vantagens

- Implementação simples.
- Distribuição uniforme de carga.

### Desvantagens

- Não considera o consumo atual de cada chave.
- Se uma chave está com problema, ainda recebe requisições.

---

## Estratégia 2 -- Baseada em consumo

Direciona as requisições para a chave com mais cota disponível:

```python
from dataclasses import dataclass, field
from datetime import date

@dataclass
class ChaveAPI:
 """Representa uma chave de API com controle de consumo."""
 key: str
 cota_mensal: int
 consumo_diario: Dict[str, int] = field(default_factory=dict)

 @property
 def consumo_mes_atual(self) -> int:
 mes = date.today().strftime("%Y-%m")
 return sum(v for k, v in self.consumo_diario.items() if k.startswith(mes))

 @property
 def cota_disponivel(self) -> int:
 return max(0, self.cota_mensal - self.consumo_mes_atual)

 def registrar_uso(self):
 hoje = date.today().isoformat()
 self.consumo_diario[hoje] = self.consumo_diario.get(hoje, 0) + 1

class ConsumoBalancer:
 """Distribui consultas priorizando a chave com mais cota disponível."""

 BASE_URL = "https://api.cpfhub.io/cpf"
 TIMEOUT = 30

 def __init__(self, chaves: List[ChaveAPI]):
 self._chaves = chaves
 self._lock = Lock()

 def _selecionar_chave(self) -> Optional[ChaveAPI]:
 with self._lock:
 disponiveis = [c for c in self._chaves if c.cota_disponivel > 0]
 if not disponiveis:
 return None
 return max(disponiveis, key=lambda c: c.cota_disponivel)

 def consultar(self, cpf: str) -> Optional[Dict]:
 chave = self._selecionar_chave()
 if not chave:
 logger.error("Nenhuma chave com cota disponivel")
 return None

 cpf_limpo = cpf.replace(".", "").replace("-", "")

 try:
 response = requests.get(
 f"{self.BASE_URL}/{cpf_limpo}",
 headers={
 "x-api-key": chave.key,
 "Accept": "application/json"
 },
 timeout=self.TIMEOUT
 )
 response.raise_for_status()
 chave.registrar_uso()
 return response.json()

 except requests.exceptions.Timeout:
 logger.error("Timeout na consulta")
 return None
 except requests.exceptions.RequestException as e:
 logger.error(f"Erro: {e}")
 return None

 def status(self) -> str:
 """Retorna status de consumo de todas as chaves."""
 linhas = ["=== Status das Chaves ==="]
 for i, chave in enumerate(self._chaves):
 linhas.append(
 f"Chave {i+1}: {chave.consumo_mes_atual}/{chave.cota_mensal} "
 f"({chave.cota_disponivel} restantes)"
 )
 return "\n".join(linhas)

# Uso
chaves = [
 ChaveAPI(key="chave_pro_1", cota_mensal=1000),
 ChaveAPI(key="chave_pro_2", cota_mensal=1000),
]

balancer = ConsumoBalancer(chaves=chaves)
resultado = balancer.consultar("12345678900")
print(balancer.status())
```

---

## Estratégia 3 -- Com fallback automático

Se a chave primária falha, automaticamente tenta a próxima. Note que a CPFHub.io não bloqueia ao atingir a cota inclusa — a API continua respondendo e cobra R$0,15 por consulta excedente. O fallback aqui é útil para erros de rede ou respostas 5xx:

```python
class FallbackBalancer:
 """Tenta chaves em sequência até obter sucesso."""

 BASE_URL = "https://api.cpfhub.io/cpf"
 TIMEOUT = 30

 def __init__(self, api_keys: List[str]):
 self._keys = api_keys

 def consultar(self, cpf: str) -> Optional[Dict]:
 cpf_limpo = cpf.replace(".", "").replace("-", "")

 for i, chave in enumerate(self._keys):
 try:
 response = requests.get(
 f"{self.BASE_URL}/{cpf_limpo}",
 headers={
 "x-api-key": chave,
 "Accept": "application/json"
 },
 timeout=self.TIMEOUT
 )

 if response.status_code == 200:
 return response.json()

 if response.status_code >= 500:
 logger.warning(f"Erro {response.status_code} na chave {i+1}. Tentando proxima...")
 continue

 # Erros 4xx nao justificam fallback entre chaves
 response.raise_for_status()

 except requests.exceptions.Timeout:
 logger.warning(f"Timeout na chave {i+1}. Tentando proxima...")
 continue
 except requests.exceptions.ConnectionError:
 logger.warning(f"Erro de conexao na chave {i+1}. Tentando proxima...")
 continue

 logger.error("Todas as chaves falharam")
 return None
```

---

## Implementação em Node.js

```javascript
const axios = require("axios");

const BASE_URL = "https://api.cpfhub.io/cpf";
const TIMEOUT_MS = 30000;

class LoadBalancer {
 constructor(apiKeys) {
 this.apiKeys = apiKeys;
 this.currentIndex = 0;
 }

 nextKey() {
 const key = this.apiKeys[this.currentIndex];
 this.currentIndex = (this.currentIndex + 1) % this.apiKeys.length;
 return key;
 }

 async consultar(cpf) {
 const cpfLimpo = cpf.replace(/\D/g, "");

 // Tenta todas as chaves como fallback
 for (let i = 0; i < this.apiKeys.length; i++) {
 const key = this.nextKey();

 try {
 const response = await axios.get(`${BASE_URL}/${cpfLimpo}`, {
 headers: {
 "x-api-key": key,
 "Accept": "application/json"
 },
 timeout: TIMEOUT_MS
 });

 return response.data;

 } catch (error) {
 const status = error.response ? error.response.status : null;
 if (status >= 500 || !status) {
 console.warn(`Chave ${i + 1} falhou (${status || "timeout"}). Tentando proxima...`);
 continue;
 }
 throw error; // Erros nao-retentaveis
 }
 }

 return null;
 }
}

// Uso
const balancer = new LoadBalancer([
 "chave_pro_1",
 "chave_pro_2",
 "chave_pro_3"
]);

balancer.consultar("12345678900").then(result => {
 if (result && result.success) {
 console.log(`Nome: ${result.data.name}`);
 }
});
```

---

## Verificando o status via cURL

Para monitorar a saúde de cada chave individualmente:

```bash
# Testar chave 1
curl -s -o /dev/null -w "Chave 1: HTTP %{http_code} em %{time_total}s\n" \
 "https://api.cpfhub.io/cpf/12345678900" \
 -H "x-api-key: CHAVE_1" \
 -H "Accept: application/json" \
 --connect-timeout 10 \
 --max-time 30

# Testar chave 2
curl -s -o /dev/null -w "Chave 2: HTTP %{http_code} em %{time_total}s\n" \
 "https://api.cpfhub.io/cpf/12345678900" \
 -H "x-api-key: CHAVE_2" \
 -H "Accept: application/json" \
 --connect-timeout 10 \
 --max-time 30
```

---

## Comparativo das estratégias

| Aspecto | Round-robin | Baseada em consumo | Com fallback |
|--------------------|--------------------|---------------------|----------------------|
| Complexidade | Baixa | Média | Baixa |
| Distribuição | Uniforme | Inteligente | Sequencial |
| Resiliência | Baixa | Média | Alta |
| Controle de cota | Nenhum | Completo | Parcial (via erros 5xx) |
| Melhor para | Volume previsível | Gestão de custos | Alta disponibilidade |

---

## Quando preferir o plano Corporativo

Se você está gerenciando mais de duas chaves Pro para atingir o volume necessário, provavelmente é hora de considerar o plano Corporativo da [**CPFHub.io**](https://www.cpfhub.io/)

---

## Perguntas frequentes

### O que é load balancing de chaves de API e por que usá-lo em consultas de CPF?
Load balancing de chaves de API é a técnica de distribuir requisições entre múltiplas credenciais de autenticação, em vez de concentrar tudo em uma única chave. Para consultas de CPF em alto volume, isso aumenta a cota efetiva, adiciona redundância e permite segregar consumo por ambiente ou departamento. A estratégia mais simples é o round-robin; a mais resiliente combina seleção por cota disponível com fallback automático para erros de rede ou instabilidades de infraestrutura.

### A CPFHub.io bloqueia quando a cota mensal é atingida?
Não. A CPFHub.io não bloqueia nem retorna erro ao atingir a cota inclusa no plano. O plano gratuito inclui 50 consultas/mês e o Pro inclui 1.000 por R$149/mês. Após o limite, a API continua respondendo normalmente e cobra R$0,15 por consulta excedente — o que simplifica a lógica de fallback, pois não há código de erro específico de cota para tratar.

### Como monitorar qual chave está consumindo mais cota?
A abordagem mais robusta é manter um contador local por chave, incrementado a cada consulta bem-sucedida. Para múltiplas instâncias da aplicação, use um contador centralizado em [Redis](https://redis.io/docs/latest/) com incremento atômico (`INCR`) e expiração automática no início de cada mês, eliminando a necessidade de lógica de reset manual.

### Quantas chaves Pro são necessárias para substituir um plano Corporativo?
Depende do volume. Cada chave Pro oferece 1.000 consultas mensais incluídas, a R$149/mês. Para 5.000 consultas/mês seriam 5 chaves (R$745/mês). A partir de 3 ou 4 chaves gerenciadas simultaneamente, vale avaliar o plano Corporativo da [CPFHub.io](https://www.cpfhub.io/) pela simplificação operacional e pelo suporte dedicado.

### Leia também

- [SLA de API de CPF: entenda os níveis de disponibilidade](https://cpfhub.io/blog/sla-api-cpf-niveis-disponibilidade)
- [Como implementar circuit breaker em integrações com API de CPF](https://cpfhub.io/blog/como-implementar-circuit-breaker-integracoes-api-cpf)
- [Como implementar retry com backoff exponencial em consultas de API de CPF](https://cpfhub.io/blog/como-implementar-retry-backoff-exponencial-consultas-api-cpf)
- [Como monitorar consumo e custos da API de CPF com observabilidade](https://cpfhub.io/blog/api-cpf-monitorar-consumo-custos-observabilidade)

---

## Conclusão

O load balancing entre múltiplas chaves de API é uma técnica poderosa para escalar consultas de CPF, adicionar redundância e gerenciar cotas de forma inteligente. A estratégia ideal depende do seu cenário: round-robin para simplicidade, baseada em consumo para controle de custos e fallback para máxima disponibilidade.

Para operações de grande escala, a [**CPFHub.io**](https://www.cpfhub.io/)

Cadastre-se em [cpfhub.io](https://www.cpfhub.io/)