Connection Pool em Python Puro: O Gerenciador Que Reutiliza Conexões HTTP e Elimina 80% do Overhead de Handshake TCP (Sem urllib3, Sem requests)
Toda requisição HTTP que você faz carrega um custo invisível: o handshake TCP. Três pacotes de ida e volta só para estabelecer a conexão antes de enviar o primeiro byte de dado real. Multiplique isso por 500 requisições em sequência e você tem 1500 pacotes desperdiçados — sem contar o TLS handshake por cima, que adiciona mais duas viagens.
Um connection pool resolve isso de forma elegante: ele mantém conexões abertas e as reutiliza. Em vez de abrir e fechar a cada request, você pega uma conexão do pool, usa, devolve, e a próxima requisição aproveita o canal já estabelecido.
Bibliotecas como urllib3 e requests fazem isso por baixo dos panos. Mas se você quer entender exatamente como funciona — ou precisa de controle total sem dependências externas — implementar do zero em Python puro é mais didático (e mais divertido) do que parece.
O Problema: 500 Requests, 500 Handshakes
Vamos medir o impacto real. Sem pool, cada requisição abre uma conexão nova:
import http.client
import time
def sem_pool(host, path, n=100):
"""Cada request abre e fecha sua própria conexão."""
tempos = []
for _ in range(n):
inicio = time.perf_counter()
conn = http.client.HTTPConnection(host, timeout=10)
conn.request("GET", path)
resp = conn.getresponse()
resp.read()
conn.close()
tempos.append(time.perf_counter() - inicio)
return tempos
tempos = sem_pool("httpbin.org", "/get", n=50)
media = sum(tempos) / len(tempos)
print(f"Sem pool: {media1000:.1f}ms por request")
print(f"Total: {sum(tempos):.2f}s para 50 requests")
No meu ambiente, isso dá ~180ms por request e ~9 segundos para 50 requisições. Agora com keep-alive reutilizando a mesma conexão:
def com_keepalive(host, path, n=100):
"""Uma única conexão reutilizada para todos os requests."""
tempos = []
conn = http.client.HTTPConnection(host, timeout=10)
for _ in range(n):
inicio = time.perf_counter()
conn.request("GET", path)
resp = conn.getresponse()
resp.read()
tempos.append(time.perf_counter() - inicio)
conn.close()
return tempos
tempos = com_keepalive("httpbin.org", "/get", n=50)
media = sum(tempos) / len(tempos)
print(f"Com keep-alive: {media1000:.1f}ms por request")
print(f"Total: {sum(tempos):.2f}s para 50 requests")
Resultado: ~95ms por request e ~4.7 segundos. Quase o dobro de velocidade só por não reabrir a conexão. Mas keep-alive simples tem um problema: uma única conexão não serve para concorrência. É aí que o pool entra.
O Pool: Múltiplas Conexões, Reutilização Inteligente
Um connection pool precisa de três coisas:
1. Criar conexões sob demanda (até um limite máximo)
2. Reutilizar conexões ociosas quando possível
3. Descartar conexões quebradas ou expiradas
Aqui está a implementação completa usando apenas a stdlib:
import http.client
import threading
import time
from collections import deque
from contextlib import contextmanager
class PooledConnection:
"""Wrapper que rastreia metadata de uma conexão no pool."""
__slots__ = ('conn', 'created_at', 'last_used', 'use_count')
def __init__(self, conn):
self.conn = conn
self.created_at = time.monotonic()
self.last_used = self.created_at
self.use_count = 0
def is_alive(self, max_age=300, max_idle=60):
"""Verifica se a conexão ainda é utilizável."""
now = time.monotonic()
if now - self.created_at > max_age:
return False
if now - self.last_used > max_idle:
return False
return True
def mark_used(self):
self.last_used = time.monotonic()
self.use_count += 1
class ConnectionPool:
"""
Pool de conexões HTTP thread-safe com reutilização inteligente.
Parâmetros:
host: hostname do servidor
port: porta (default 80 para HTTP, 443 para HTTPS)
max_connections: tamanho máximo do pool
max_age: tempo máximo de vida de uma conexão (segundos)
max_idle: tempo máximo ociosa antes de descartar (segundos)
use_ssl: usar HTTPS
timeout: timeout por conexão (segundos)
"""
def __init__(self, host, port=None, max_connections=10, max_age=300,
max_idle=60, use_ssl=False, timeout=10):
self.host = host
self.port = port or (443 if use_ssl else 80)
self.max_connections = max_connections
self.max_age = max_age
self.max_idle = max_idle
self.use_ssl = use_ssl
self.timeout = timeout
self._pool = deque() # conexões ociosas disponíveis
self._active = 0 # conexões em uso
self._lock = threading.Lock()
self._semaphore = threading.Semaphore(max_connections)
# Estatísticas
self._stats = {
'created': 0,
'reused': 0,
'discarded': 0,
'errors': 0,
}
def _create_connection(self):
"""Cria uma nova conexão HTTP/HTTPS."""
if self.use_ssl:
conn = http.client.HTTPSConnection(
self.host, self.port, timeout=self.timeout
)
else:
conn = http.client.HTTPConnection(
self.host, self.port, timeout=self.timeout
)
self._stats['created'] += 1
return PooledConnection(conn)
def _is_healthy(self, pooled_conn):
"""Testa se a conexão ainda responde."""
if not pooled_conn.is_alive(self.max_age, self.max_idle):
return False
try:
# Teste leve: verifica se o socket ainda está aberto
sock = pooled_conn.conn.sock
if sock is None:
return False
# Se o servidor fechou, recv retorna vazio
sock.setblocking(False)
try:
data = sock.recv(1)
if data == b'':
return False # conexão fechada pelo servidor
except BlockingIOError:
pass # sem dados = conexão aberta (bom sinal)
except (ConnectionError, OSError):
return False
finally:
sock.setblocking(True)
return True
except Exception:
return False
def acquire(self):
"""Obtém uma conexão do pool (cria se necessário)."""
self._semaphore.acquire()
with self._lock:
# Tenta reutilizar uma conexão ociosa
while self._pool:
pooled = self._pool.popleft()
if self._is_healthy(pooled):
pooled.mark_used()
self._active += 1
self._stats['reused'] += 1
return pooled
else:
# Conexão morta — descarta
self._close_silent(pooled)
self._stats['discarded'] += 1
# Pool vazio — cria nova conexão
pooled = self._create_connection()
pooled.mark_used()
self._active += 1
return pooled
def release(self, pooled_conn, broken=False):
"""Devolve uma conexão ao pool (ou descarta se quebrada)."""
with self._lock:
self._active -= 1
if broken or not self._is_healthy(pooled_conn):
self._close_silent(pooled_conn)
self._stats['discarded'] += 1
else:
self._pool.append(pooled_conn)
self._semaphore.release()
def _close_silent(self, pooled_conn):
"""Fecha conexão ignorando exceções."""
try:
pooled_conn.conn.close()
except Exception:
pass
@contextmanager
def connection(self):
"""
Context manager para uso seguro.
Uso:
with pool.connection() as conn:
conn.request("GET", "/api")
resp = conn.getresponse()
"""
pooled = self.acquire()
broken = False
try:
yield pooled.conn
except (ConnectionError, OSError, http.client.HTTPException):
broken = True
self._stats['errors'] += 1
raise
finally:
self.release(pooled, broken=broken)
def drain(self):
"""Fecha todas as conexões ociosas. Chame no shutdown."""
with self._lock:
while self._pool:
pooled = self._pool.popleft()
self._close_silent(pooled)
self._stats['discarded'] += 1
@property
def stats(self):
with self._lock:
return {
self._stats,
'idle': len(self._pool),
'active': self._active,
'total': len(self._pool) + self._active,
}
Uso Real: Pool com Threads
O pool brilha quando múltiplas threads precisam fazer requests concorrentes:
import concurrent.futures
pool = ConnectionPool("httpbin.org", max_connections=10, use_ssl=True)
def fazer_request(i):
with pool.connection() as conn:
conn.request("GET", f"/delay/1")
resp = conn.getresponse()
data = resp.read()
return i, resp.status
# 30 requests com no máximo 10 conexões simultâneas
with concurrent.futures.ThreadPoolExecutor(max_workers=10) as executor:
futures = [executor.submit(fazer_request, i) for i in range(30)]
for f in concurrent.futures.as_completed(futures):
idx, status = f.result()
print(f"Request {idx}: HTTP {status}")
print(f"\nEstatísticas do pool:")
for k, v in pool.stats.items():
print(f" {k}: {v}")
pool.drain()
Com 10 conexões no pool e 30 requests que levam ~1s cada (o endpoint /delay/1 do httpbin), tudo termina em ~3 segundos. Sem o pool, seriam 30 conexões abertas simultaneamente — e muitos servidores limitam isso.
Health Check Passivo vs Ativo
O pool acima usa health check passivo: testa a conexão quando tenta reutilizá-la. Para sistemas críticos, você pode adicionar um health check ativo* que testa conexões periodicamente em background:
class ConnectionPoolWithHealthCheck(ConnectionPool):
"""Pool com thread de health check periódico."""
def __init__(self, args, health_interval=30, *kwargs):
super().__init__(args, kwargs)
self._health_interval = health_interval
self._running = True
self._health_thread = threading.Thread(
target=self._health_loop, daemon=True
)
self._health_thread.start()
def _health_loop(self):
"""Testa conexões ociosas periodicamente."""
while self._running:
time.sleep(self._health_interval)
with self._lock:
saudaveis = deque()
while self._pool:
pooled = self._pool.popleft()
if self._is_healthy(pooled):
saudaveis.append(pooled)
else:
self._close_silent(pooled)
self._stats['discarded'] += 1
self._pool = saudaveis
def drain(self):
self._running = False
super().drain()
Benchmarks: Sem Pool vs Pool vs requests
Para 100 requests ao httpbin.org/get:
| Método | Tempo total | Conexões criadas |
|---|---|---|
| Sem pool (novo por request) | ~18s | 100 |
| Keep-alive simples | ~9s | 1 |
| ConnectionPool (10 conexões) | ~2s | 10 |
requests com Session |
~2s | ~10 |
O pool caseiro performa comparável ao requests com Session — mas você tem controle total sobre timeouts, limites e estratégia de descarte.
O Perrengue do Olivetto
Confesso: a primeira versão desse pool tinha um bug sutil que me consumiu 4 horas de debugging. O sock.recv(1) no health check consumia um byte do buffer da conexão. Se o servidor tivesse enviado dados (como um chunked encoding trailer), o próximo request via aquela conexão recebia dados corrompidos.
O erro se manifestava como um http.client.BadStatusLine intermitente que só aparecia sob carga. Testes unitários passavam. Em produção, a cada ~200 requests, um falhava com uma resposta garbage.
A solução foi envolver o recv num bloco setblocking(False) e verificar se retorna BlockingIOError (conexão viva, sem dados) ou bytes vazios (conexão fechada). Mas o susto ficou: health check de socket é terra de ninguém e qualquer byte consumido é um byte que o application layer não espera.
Se você implementar health check passivo, considere simplesmente tentar o request e fazer retry com conexão nova se falhar. É menos elegante, mas não rouba bytes do buffer:
@contextmanager
def connection_with_retry(self, max_retries=1):
"""Tenta o request; se a conexão estiver morta, pega outra."""
for attempt in range(max_retries + 1):
pooled = self.acquire()
try:
yield pooled.conn
self.release(pooled)
return
except (ConnectionError, OSError, http.client.HTTPException):
self.release(pooled, broken=True)
if attempt == max_retries:
raise
Quando NÃO Usar Pool
Connection pool não é bala de prata. Evite quando:
- Servidores diferentes a cada request — o pool é por host/port. Se você bate em 50 hosts diferentes, precisa de 50 pools
- Requests raros — se você faz 1 request a cada 5 minutos, a conexão vai expirar entre usos de qualquer forma
- Protocolo não suporta keep-alive — servidores que respondem com
Connection: closeforçam o fechamento. O pool vai criar e descartar conexões sem benefício real - WebSockets ou streaming — conexões de longa duração não devem ir para o pool. Elas bloqueiam um slot permanentemente
O Que Fazer Depois
O pool acima cobre 90% dos casos. Para ir além, considere:
- Stale checking com TCP_NODELAY — detecta conexões mortas mais rápido
- Priority queue — connections que tiveram menos erros são preferidas
- Circuit breaker integrado** — se um host começa a falhar, pare de pegar conexões do pool dele
E você? Já teve que debugar um problema de conexão que só aparecia em produção sob carga? Conta nos comentários — especialmente se foi um daqueles bugs que some quando você tenta reproduzir.
