Quando si effettuano il parsing di migliaia di pagine di marketplace, si inviano richieste API in massa o si automatizza il lavoro con centinaia di account, una corretta distribuzione del carico tramite proxy diventa fondamentale. Senza un bilanciamento adeguato, ci si può imbattere in blocchi, timeout e cali di prestazioni. In questa guida esamineremo l'architettura della distribuzione del carico, gli algoritmi di bilanciamento e le strategie pratiche per sistemi ad alto carico.
Il materiale è destinato a sviluppatori e specialisti tecnici che lavorano con il parsing dei dati, l'automazione delle richieste API o gestiscono grandi pool di proxy per esigenze aziendali.
Perché è necessario il bilanciamento del carico tramite proxy
Il bilanciamento del carico tramite proxy risolve diverse problematiche critiche nei sistemi ad alto carico. La prima e principale è la protezione dai blocchi. Quando si inviano migliaia di richieste a una risorsa (marketplace, API di social media, motore di ricerca), il server di destinazione rileva un'attività anormalmente alta da un singolo indirizzo IP e lo blocca. Distribuire le richieste tra decine o centinaia di proxy rende la propria attività simile a quella di utenti normali.
La seconda problematica è la performance. Un singolo server proxy ha una larghezza di banda limitata (di solito 100-1000 Mbit/s) e può gestire un numero limitato di connessioni simultanee. Durante il parsing di 10.000 pagine al minuto o l'invio massivo di richieste API, un singolo proxy diventa un collo di bottiglia nel sistema. Il bilanciamento consente di scalare orizzontalmente la larghezza di banda, aggiungendo nuovi proxy al pool.
La terza questione è l'affidabilità. Se uno dei proxy si guasta (guasto tecnico, blocco, scadenza della locazione), il sistema reindirizza automaticamente il traffico ai proxy funzionanti. Senza un meccanismo di bilanciamento e controlli di salute, il guasto di un proxy può fermare l'intero sistema.
Esempio reale: Durante il parsing di Wildberries per monitorare i prezzi dei concorrenti, è necessario elaborare 50.000 prodotti ogni ora. Questo corrisponde a circa 14 richieste al secondo. Un singolo proxy può gestire tale carico, ma Wildberries bloccherà l'IP dopo 100-200 richieste da un singolo indirizzo. Distribuendo le richieste tra 20 proxy residenziali, si riduce il carico su ogni IP a 0.7 richieste al secondo — questo appare come l'attività di un utente normale.
La quarta ragione è la distribuzione geografica. Molti servizi mostrano contenuti o prezzi diversi a seconda della regione dell'utente. Il bilanciamento tra proxy di diversi paesi e città consente di raccogliere dati da tutte le regioni target contemporaneamente. Ad esempio, per monitorare i prezzi su Ozon, sono necessari proxy da Mosca, San Pietroburgo, Ekaterinburg e altre grandi città.
Architettura del sistema di distribuzione del carico
L'architettura classica di un sistema di bilanciamento del carico tramite proxy è composta da diversi componenti. Al livello superiore si trova il bilanciatore (load balancer) — un modulo software che riceve i compiti in ingresso (richieste di parsing, chiamate API) e li distribuisce tra i proxy disponibili. Il bilanciatore può funzionare come un servizio separato o essere integrato nell'applicazione.
Il secondo componente è il gestore del pool di proxy (proxy pool manager). Esso memorizza l'elenco di tutti i proxy disponibili con le loro caratteristiche: indirizzo IP, porta, protocollo (HTTP/SOCKS5), posizione geografica, tipo (residenziale, mobile, data center), stato attuale (attivo, bloccato, in verifica). Il gestore del pool è responsabile dell'aggiunta di nuovi proxy, della rimozione di quelli non funzionanti e della verifica periodica della disponibilità.
Il terzo elemento è il sistema di monitoraggio e controlli di salute. Esso verifica costantemente la funzionalità di ogni proxy: invia richieste di test, misura i tempi di risposta, verifica il successo della connessione. Se un proxy non risponde o restituisce errori, il sistema lo contrassegna come non disponibile ed esclude temporaneamente dalla rotazione.
| Componente | Funzione | Tecnologie |
|---|---|---|
| Load Balancer | Distribuzione delle richieste tra i proxy | HAProxy, Nginx, librerie Python/Node.js |
| Proxy Pool Manager | Gestione dell'elenco dei proxy | Redis, PostgreSQL, memorie in-memory |
| Health Check System | Verifica della disponibilità dei proxy | Task programmati, Celery, cron |
| Rate Limiter | Controllo della frequenza delle richieste | Algoritmi Token bucket, leaky bucket |
| Monitoring & Metrics | Raccolta di metriche delle prestazioni | Prometheus, Grafana, ELK Stack |
Il quarto componente è il rate limiter (limitatore di frequenza delle richieste). Esso si assicura che ogni proxy non superi la frequenza consentita di accesso alla risorsa di destinazione. Ad esempio, se si effettua il parsing di Instagram e si sa che la piattaforma blocca l'IP dopo 60 richieste al minuto, il rate limiter non permetterà di inviare più di 50 richieste al minuto attraverso un singolo proxy, lasciando un margine di sicurezza.
Il quinto elemento è il sistema di metriche e analisi. Esso raccoglie dati sulle prestazioni di ogni proxy: numero di richieste riuscite, numero di errori, tempo medio di risposta, percentuale di blocchi. Questi dati vengono utilizzati per ottimizzare gli algoritmi di bilanciamento e identificare i proxy problematici.
Algoritmi di bilanciamento: Round Robin, Least Connections, Weighted
L'algoritmo Round Robin è il metodo di bilanciamento più semplice e comune. Esso scorre sequenzialmente i proxy dall'elenco: la prima richiesta passa attraverso il proxy n. 1, la seconda attraverso il proxy n. 2, la terza attraverso il proxy n. 3, e così via. Quando l'elenco finisce, il bilanciatore torna all'inizio. Questo algoritmo distribuisce uniformemente il carico, se tutti i proxy hanno prestazioni simili.
class RoundRobinBalancer:
def __init__(self, proxies):
self.proxies = proxies
self.current_index = 0
def get_next_proxy(self):
proxy = self.proxies[self.current_index]
self.current_index = (self.current_index + 1) % len(self.proxies)
return proxy
# Utilizzo
proxies = [
"http://proxy1.example.com:8080",
"http://proxy2.example.com:8080",
"http://proxy3.example.com:8080"
]
balancer = RoundRobinBalancer(proxies)
for i in range(10):
proxy = balancer.get_next_proxy()
print(f"Richiesta {i+1} → {proxy}")L'algoritmo Least Connections (minore numero di connessioni) sceglie il proxy con il minor numero di connessioni attive in quel momento. Questo è utile quando le richieste hanno durate di esecuzione diverse. Ad esempio, durante il parsing, una richiesta può essere elaborata in 100 ms, mentre un'altra in 5 secondi (se la pagina si carica lentamente). Least Connections reindirizza automaticamente le nuove richieste ai proxy meno occupati.
class LeastConnectionsBalancer:
def __init__(self, proxies):
self.proxies = {proxy: 0 for proxy in proxies}
def get_next_proxy(self):
# Scegliamo il proxy con il minor numero di connessioni
proxy = min(self.proxies.items(), key=lambda x: x[1])[0]
self.proxies[proxy] += 1
return proxy
def release_proxy(self, proxy):
# Riduciamo il contatore al termine della richiesta
self.proxies[proxy] -= 1
# Utilizzo con il gestore di contesto
balancer = LeastConnectionsBalancer(proxies)
def make_request(url):
proxy = balancer.get_next_proxy()
try:
# Eseguiamo la richiesta attraverso il proxy selezionato
response = requests.get(url, proxies={"http": proxy})
return response
finally:
balancer.release_proxy(proxy)Weighted Round Robin (round robin pesato) estende il classico Round Robin, assegnando a ciascun proxy un peso in base alle sue prestazioni o qualità. I proxy con un peso maggiore ricevono più richieste. Questo è utile quando si hanno proxy di qualità diversa: ad esempio, proxy residenziali premium con alta velocità e proxy economici di data center con limitazioni.
class WeightedRoundRobinBalancer:
def __init__(self, weighted_proxies):
# weighted_proxies = [(proxy, weight), ...]
self.proxies = []
for proxy, weight in weighted_proxies:
# Aggiungiamo il proxy alla lista weight volte
self.proxies.extend([proxy] * weight)
self.current_index = 0
def get_next_proxy(self):
proxy = self.proxies[self.current_index]
self.current_index = (self.current_index + 1) % len(self.proxies)
return proxy
# Utilizzo con pesi
weighted_proxies = [
("http://premium-proxy1.com:8080", 5), # alta qualità
("http://premium-proxy2.com:8080", 5),
("http://cheap-proxy1.com:8080", 2), # bassa qualità
("http://cheap-proxy2.com:8080", 1) # qualità molto bassa
]
balancer = WeightedRoundRobinBalancer(weighted_proxies)L'algoritmo Random (scelta casuale) seleziona un proxy casualmente dall'elenco disponibile. È più semplice da implementare rispetto al Round Robin e funziona bene con un gran numero di proxy (100+). La distribuzione casuale si bilancia automaticamente con un volume sufficiente di richieste. Lo svantaggio è che possono verificarsi brevi periodi di carico non uniforme.
IP Hash è un algoritmo che seleziona un proxy sulla base dell'hash dell'URL di destinazione o di un altro parametro. Questo garantisce che le richieste a una stessa risorsa passino sempre attraverso lo stesso proxy. È utile per siti che utilizzano sessioni o richiedono una sequenza di richieste da un singolo IP (ad esempio, autenticazione + recupero dati).
Gestione dei pool di proxy: rotazione e controlli di salute
Una gestione efficace del pool di proxy richiede un monitoraggio costante dello stato di ogni proxy e una rotazione automatica. Il controllo di salute è una verifica periodica della disponibilità del proxy tramite l'invio di una richiesta di test. Di solito si utilizza una semplice richiesta GET a un servizio affidabile (ad esempio, httpbin.org o un proprio endpoint) che restituisce informazioni sull'indirizzo IP.
import requests
import time
from datetime import datetime
class ProxyHealthChecker:
def __init__(self, test_url="http://httpbin.org/ip", timeout=10):
self.test_url = test_url
self.timeout = timeout
def check_proxy(self, proxy_url):
"""Controlla la funzionalità del proxy"""
try:
start_time = time.time()
response = requests.get(
self.test_url,
proxies={"http": proxy_url, "https": proxy_url},
timeout=self.timeout
)
response_time = time.time() - start_time
if response.status_code == 200:
return {
"status": "healthy",
"response_time": response_time,
"timestamp": datetime.now(),
"ip": response.json().get("origin")
}
else:
return {
"status": "unhealthy",
"error": f"HTTP {response.status_code}",
"timestamp": datetime.now()
}
except requests.exceptions.Timeout:
return {
"status": "unhealthy",
"error": "timeout",
"timestamp": datetime.now()
}
except Exception as e:
return {
"status": "unhealthy",
"error": str(e),
"timestamp": datetime.now()
}
# Utilizzo
checker = ProxyHealthChecker()
proxies = ["http://proxy1.com:8080", "http://proxy2.com:8080"]
for proxy in proxies:
result = checker.check_proxy(proxy)
print(f"{proxy}: {result['status']} ({result.get('response_time', 'N/A')}s)")Il sistema di gestione del pool deve automaticamente contrassegnare i proxy non funzionanti come non disponibili e ripetere periodicamente il controllo. Una strategia comune è il pattern del circuit breaker: dopo tre controlli consecutivi non riusciti, il proxy viene escluso dal pool per 5-10 minuti, quindi viene controllato nuovamente. Se il controllo ha successo, il proxy torna nel pool attivo.
class ProxyPoolManager:
def __init__(self, health_checker, max_failures=3, cooldown_seconds=300):
self.health_checker = health_checker
self.max_failures = max_failures
self.cooldown_seconds = cooldown_seconds
self.proxies = {} # {proxy_url: ProxyInfo}
def add_proxy(self, proxy_url, metadata=None):
"""Aggiunge un proxy al pool"""
self.proxies[proxy_url] = {
"url": proxy_url,
"status": "active",
"failures": 0,
"last_check": None,
"cooldown_until": None,
"metadata": metadata or {}
}
def get_active_proxies(self):
"""Restituisce l'elenco dei proxy attivi"""
now = datetime.now()
active = []
for proxy_url, info in self.proxies.items():
# Controlliamo se il proxy è in cooldown
if info["cooldown_until"] and now < info["cooldown_until"]:
continue
if info["status"] == "active":
active.append(proxy_url)
return active
def mark_failure(self, proxy_url):
"""Segna un tentativo di utilizzo del proxy come non riuscito"""
if proxy_url not in self.proxies:
return
info = self.proxies[proxy_url]
info["failures"] += 1
if info["failures"] >= self.max_failures:
# Passa in cooldown
info["status"] = "cooldown"
info["cooldown_until"] = datetime.now() + timedelta(seconds=self.cooldown_seconds)
print(f"Proxy {proxy_url} spostato in cooldown fino a {info['cooldown_until']}")
def mark_success(self, proxy_url):
"""Segna un utilizzo del proxy come riuscito"""
if proxy_url not in self.proxies:
return
info = self.proxies[proxy_url]
info["failures"] = 0
info["status"] = "active"
info["cooldown_until"] = NoneLa rotazione dei proxy è una strategia di cambio automatico dei proxy a intervalli definiti o dopo un certo numero di richieste. Esistono diversi approcci: rotazione temporale (cambio ogni 5-10 minuti), rotazione per numero di richieste (cambio dopo 100-500 richieste), rotazione per sessioni (un proxy per una sessione di parsing). La scelta della strategia dipende dai requisiti del sito target.
Consiglio: Per il parsing di marketplace (Wildberries, Ozon) la rotazione per numero di richieste è ottimale: 50-100 richieste per un proxy, poi cambio. Per lavorare con API di social media (Instagram, Facebook) è meglio utilizzare la rotazione per sessioni: un proxy per l'intero ciclo di autenticazione → azioni → uscita.
Rate limiting e controllo della frequenza delle richieste
Il rate limiting è un componente critico del sistema di bilanciamento del carico. Esso previene il superamento della frequenza consentita di richieste alla risorsa di destinazione, il che potrebbe portare al blocco dei proxy. Esistono due algoritmi principali: Token Bucket (secchio di token) e Leaky Bucket (secchio forato).
Il Token Bucket funziona nel seguente modo: ogni proxy ha un "secchio" virtuale di token. Ogni token consente una richiesta. Il secchio si riempie gradualmente di token a una velocità prestabilita (ad esempio, 10 token al secondo). La capacità massima del secchio è limitata (ad esempio, 50 token). Quando arriva una richiesta, il sistema verifica la disponibilità di token: se ci sono — ne preleva uno e esegue la richiesta, se non ci sono — attende l'arrivo di un nuovo token.
import time
from threading import Lock
class TokenBucketRateLimiter:
def __init__(self, rate, capacity):
"""
rate: numero di token al secondo
capacity: numero massimo di token nel secchio
"""
self.rate = rate
self.capacity = capacity
self.tokens = capacity
self.last_update = time.time()
self.lock = Lock()
def _add_tokens(self):
"""Aggiunge token in base al tempo trascorso"""
now = time.time()
elapsed = now - self.last_update
new_tokens = elapsed * self.rate
self.tokens = min(self.capacity, self.tokens + new_tokens)
self.last_update = now
def acquire(self, tokens=1):
"""Cerca di ottenere il numero specificato di token"""
with self.lock:
self._add_tokens()
if self.tokens >= tokens:
self.tokens -= tokens
return True
return False
def wait_and_acquire(self, tokens=1):
"""Attende l'arrivo di token e li ottiene"""
while not self.acquire(tokens):
# Calcola il tempo di attesa
wait_time = (tokens - self.tokens) / self.rate
time.sleep(wait_time)
# Utilizzo per ogni proxy
proxy_limiters = {
"http://proxy1.com:8080": TokenBucketRateLimiter(rate=10, capacity=50),
"http://proxy2.com:8080": TokenBucketRateLimiter(rate=10, capacity=50)
}
def make_request_with_limit(url, proxy_url):
limiter = proxy_limiters[proxy_url]
limiter.wait_and_acquire() # Attende la disponibilità del token
response = requests.get(url, proxies={"http": proxy_url})
return responseIl Leaky Bucket (secchio forato) funziona in modo diverso: le richieste vengono inserite in una coda (secchio), che "fuoriesce" a una velocità costante. Se il secchio trabocca, le nuove richieste vengono rifiutate o messe in attesa. Questo algoritmo garantisce una distribuzione del carico più uniforme nel tempo, ma può portare a ritardi durante i picchi di attività.
Diverse piattaforme target richiedono limiti diversi. L'API di Instagram consente circa 200 richieste all'ora per un singolo IP (circa 3.3 richieste al minuto). Wildberries blocca dopo 100-200 richieste al minuto da un singolo IP. Google Search consente 10-20 richieste al minuto. Il vostro sistema di rate limiting deve tenere conto di queste limitazioni e aggiungere un margine di sicurezza (di solito 20-30%).
| Piattaforma | Limite di richieste | Impostazione consigliata |
|---|---|---|
| ~200 richieste/ora | 2-3 richieste/minuto (con margine) | |
| Wildberries | 100-200 richieste/minuto | 60-80 richieste/minuto |
| Google Search | 10-20 richieste/minuto | 8-12 richieste/minuto |
| Ozon | 50-100 richieste/minuto | 30-50 richieste/minuto |
| Facebook API | 200 richieste/ora | 2-3 richieste/minuto |
Monitoraggio delle prestazioni e scalabilità automatica
Un sistema di bilanciamento del carico efficace richiede un monitoraggio costante delle metriche chiave. Il primo gruppo di metriche riguarda le prestazioni dei proxy: tempo medio di risposta (response time), percentuale di richieste riuscite (success rate), numero di timeout, numero di errori 4xx e 5xx. Questi dati consentono di identificare i proxy problematici e ottimizzare gli algoritmi di bilanciamento.
class ProxyMetrics:
def __init__(self):
self.metrics = {} # {proxy_url: metrics_dict}
def record_request(self, proxy_url, response_time, status_code, success):
"""Registra le metriche della richiesta"""
if proxy_url not in self.metrics:
self.metrics[proxy_url] = {
"total_requests": 0,
"successful_requests": 0,
"failed_requests": 0,
"total_response_time": 0,
"timeouts": 0,
"errors_4xx": 0,
"errors_5xx": 0
}
m = self.metrics[proxy_url]
m["total_requests"] += 1
if success:
m["successful_requests"] += 1
m["total_response_time"] += response_time
else:
m["failed_requests"] += 1
if status_code == 0: # timeout
m["timeouts"] += 1
elif 400 <= status_code < 500:
m["errors_4xx"] += 1
elif 500 <= status_code < 600:
m["errors_5xx"] += 1
def get_success_rate(self, proxy_url):
"""Restituisce la percentuale di richieste riuscite"""
m = self.metrics.get(proxy_url, {})
total = m.get("total_requests", 0)
if total == 0:
return 0
return (m.get("successful_requests", 0) / total) * 100
def get_avg_response_time(self, proxy_url):
"""Restituisce il tempo medio di risposta"""
m = self.metrics.get(proxy_url, {})
successful = m.get("successful_requests", 0)
if successful == 0:
return 0
return m.get("total_response_time", 0) / successful
def get_report(self, proxy_url):
"""Restituisce un rapporto completo sul proxy"""
m = self.metrics.get(proxy_url, {})
return {
"proxy": proxy_url,
"total_requests": m.get("total_requests", 0),
"success_rate": self.get_success_rate(proxy_url),
"avg_response_time": self.get_avg_response_time(proxy_url),
"timeouts": m.get("timeouts", 0),
"errors_4xx": m.get("errors_4xx", 0),
"errors_5xx": m.get("errors_5xx", 0)
}Il secondo gruppo di metriche riguarda le prestazioni complessive del sistema: numero totale di richieste al secondo (RPS — requests per second), latenza media (latency), dimensione della coda delle richieste, numero di proxy attivi, percentuale di utilizzo del pool di proxy. Queste metriche mostrano se il sistema sta gestendo il carico o se è necessaria la scalabilità.
La scalabilità automatica (auto-scaling) consente di aggiungere o rimuovere proxy dinamicamente in base al carico. Una strategia semplice: se il carico medio del pool supera l'80% per 5 minuti, il sistema aggiunge automaticamente nuovi proxy. Se il carico scende sotto il 30% per 15 minuti, il sistema rimuove i proxy in eccesso per risparmiare risorse.
Esempio di configurazione del monitoraggio: Per il parsing di 100.000 prodotti Wildberries all'ora (circa 28 richieste al secondo), sono necessari almeno 30-40 proxy con un limite di 60 richieste al minuto per proxy. Configura avvisi: se il success rate scende sotto l'85% o il tempo medio di risposta supera i 3 secondi, il sistema deve automaticamente aggiungere 10-15 proxy di riserva dal pool.
Per visualizzare le metriche, utilizza Grafana con Prometheus o ELK Stack. Crea dashboard con grafici: RPS nel tempo, distribuzione del tempo di risposta, top 10 proxy più veloci/lenti, mappa degli errori per tipo. Questo permetterà di identificare rapidamente i problemi e ottimizzare il sistema.
Implementazione pratica in Python e Node.js
Esaminiamo un'implementazione completa di un sistema di bilanciamento del carico in Python utilizzando librerie popolari. Questo esempio unisce tutti i componenti descritti: bilanciatore, gestore del pool, controlli di salute, rate limiting e monitoraggio.
import requests
import time
import random
from datetime import datetime, timedelta
from threading import Lock, Thread
from collections import defaultdict
class ProxyLoadBalancer:
def __init__(self, proxies, algorithm="round_robin", rate_limit=10):
"""
proxies: lista di proxy [{"url": "...", "weight": 1}, ...]
algorithm: round_robin, least_connections, weighted, random
rate_limit: numero massimo di richieste al secondo per proxy
"""
self.proxies = proxies
self.algorithm = algorithm
self.rate_limit = rate_limit
# Stato del bilanciatore
self.current_index = 0
self.connections = defaultdict(int)
self.rate_limiters = {}
self.metrics = defaultdict(lambda: {
"total": 0, "success": 0, "failed": 0,
"response_times": [], "last_check": None
})
# Inizializzazione dei rate limiters
for proxy in proxies:
proxy_url = proxy["url"]
self.rate_limiters[proxy_url] = TokenBucketRateLimiter(
rate=rate_limit,
capacity=rate_limit * 5
)
self.lock = Lock()
# Avvio del controllo di salute in background
self.health_check_thread = Thread(target=self._health_check_loop, daemon=True)
self.health_check_thread.start()
def get_next_proxy(self):
"""Seleziona il prossimo proxy secondo l'algoritmo"""
with self.lock:
if self.algorithm == "round_robin":
return self._round_robin()
elif self.algorithm == "least_connections":
return self._least_connections()
elif self.algorithm == "weighted":
return self._weighted_random()
elif self.algorithm == "random":
return random.choice(self.proxies)["url"]
def _round_robin(self):
proxy = self.proxies[self.current_index]["url"]
self.current_index = (self.current_index + 1) % len(self.proxies)
return proxy
def _least_connections(self):
min_conn = min(self.connections.values()) if self.connections else 0
candidates = [p["url"] for p in self.proxies if self.connections[p["url"]] == min_conn]
return random.choice(candidates)
def _weighted_random(self):
weights = [p.get("weight", 1) for p in self.proxies]
return random.choices(self.proxies, weights=weights)[0]["url"]
def make_request(self, url, method="GET", **kwargs):
"""Esegue una richiesta tramite il bilanciatore"""
proxy_url = self.get_next_proxy()
# Attende la disponibilità del rate limit
self.rate_limiters[proxy_url].wait_and_acquire()
# Incrementa il contatore delle connessioni
with self.lock:
self.connections[proxy_url] += 1
try:
start_time = time.time()
response = requests.request(
method,
url,
proxies={"http": proxy_url, "https": proxy_url},
timeout=kwargs.get("timeout", 30),
**kwargs
)
response_time = time.time() - start_time
# Registra le metriche
self._record_metrics(proxy_url, response_time, True, response.status_code)
return response
except Exception as e:
self._record_metrics(proxy_url, 0, False, 0)
raise
finally:
# Riduce il contatore delle connessioni
with self.lock:
self.connections[proxy_url] -= 1
def _record_metrics(self, proxy_url, response_time, success, status_code):
"""Registra le metriche della richiesta"""
with self.lock:
m = self.metrics[proxy_url]
m["total"] += 1
if success:
m["success"] += 1
m["response_times"].append(response_time)
# Memorizza solo gli ultimi 1000 valori
if len(m["response_times"]) > 1000:
m["response_times"].pop(0)
else:
m["failed"] += 1
def _health_check_loop(self):
"""Controllo di salute in background dei proxy"""
while True:
for proxy in self.proxies:
proxy_url = proxy["url"]
try:
response = requests.get(
"http://httpbin.org/ip",
proxies={"http": proxy_url},
timeout=10
)
with self.lock:
self.metrics[proxy_url]["last_check"] = datetime.now()
proxy["status"] = "healthy" if response.status_code == 200 else "unhealthy"
except:
with self.lock:
proxy["status"] = "unhealthy"
time.sleep(60) # Controllo ogni minuto
def get_stats(self):
"""Restituisce le statistiche su tutti i proxy"""
stats = []
with self.lock:
for proxy in self.proxies:
proxy_url = proxy["url"]
m = self.metrics[proxy_url]
avg_response_time = (
sum(m["response_times"]) / len(m["response_times"])
if m["response_times"] else 0
)
success_rate = (
(m["success"] / m["total"] * 100)
if m["total"] > 0 else 0
)
stats.append({
"proxy": proxy_url,
"status": proxy.get("status", "unknown"),
"total_requests": m["total"],
"success_rate": round(success_rate, 2),
"avg_response_time": round(avg_response_time, 3),
"active_connections": self.connections[proxy_url]
})
return statsUtilizzo di questo bilanciatore per il parsing di un marketplace:
# Configurazione del bilanciatore
proxies = [
{"url": "http://proxy1.example.com:8080", "weight": 5},
{"url": "http://proxy2.example.com:8080", "weight": 5},
{"url": "http://proxy3.example.com:8080", "weight": 3},
{"url": "http://proxy4.example.com:8080", "weight": 2}
]
balancer = ProxyLoadBalancer(
proxies=proxies,
algorithm="weighted",
rate_limit=60 # 60 richieste al minuto per proxy
)
# Parsing dell'elenco dei prodotti
product_urls = [f"https://www.wildberries.ru/catalog/{i}/detail.aspx" for i in range(1000)]
results = []
for url in product_urls:
try:
response = balancer.make_request(url)
# Elaborazione della risposta
results.append({"url": url, "status": "success", "data": response.text})
except Exception as e:
results.append({"url": url, "status": "error", "error": str(e)})
# Ogni 100 richieste, stampa le statistiche
if len(results) % 100 == 0:
stats = balancer.get_stats()
for stat in stats:
print(f"{stat['proxy']}: {stat['success_rate']}% successo, "
f"{stat['avg_response_time']}s tempo medio di risposta")
# Statistiche finali
print("\n=== Statistiche Finali ===")
for stat in balancer.get_stats():
print(f"{stat['proxy']}:")
print(f" Richieste totali: {stat['total_requests']}")
print(f" Percentuale di successo: {stat['success_rate']}%")
print(f" Tempo medio di risposta: {stat['avg_response_time']}s")
print(f" Stato: {stat['status']}")Per Node.js è possibile utilizzare un'architettura simile con le librerie axios per le richieste HTTP e node-rate-limiter per il controllo della frequenza:
const axios = require('axios');
const { RateLimiter } = require('limiter');
class ProxyLoadBalancer {
constructor(proxies, algorithm = 'round_robin', rateLimit = 10) {
this.proxies = proxies;
this.algorithm = algorithm;
this.currentIndex = 0;
this.connections = new Map();
this.limiters = new Map();
this.metrics = new Map();
// Inizializzazione dei rate limiters
proxies.forEach(proxy => {
this.limiters.set(proxy.url, new RateLimiter({
tokensPerInterval: rateLimit,
interval: 'second'
}));
this.connections.set(proxy.url, 0);
this.metrics.set(proxy.url, {
total: 0,
success: 0,
failed: 0,
responseTimes: []
});
});
}
getNextProxy() {
if (this.algorithm === 'round_robin') {
const proxy = this.proxies[this.currentIndex].url;
this.currentIndex = (this.currentIndex + 1) % this.proxies.length;
return proxy;
} else if (this.algorithm === 'least_connections') {
let minConn = Infinity;
let selectedProxy = null;
this.connections.forEach((count, proxy) => {
if (count < minConn) {
minConn = count;
selectedProxy = proxy;
}
});
return selectedProxy;
}
}
async makeRequest(url, options = {}) {
const proxyUrl = this.getNextProxy();
const limiter = this.limiters.get(proxyUrl);
// Attende la disponibilità del rate limit
await limiter.removeTokens(1);
// Incrementa il contatore delle connessioni
this.connections.set(proxyUrl, this.connections.get(proxyUrl) + 1);
try {
const startTime = Date.now();
const response = await axios({
url,
proxy: this.parseProxyUrl(proxyUrl),
timeout: options.timeout || 30000,
...options
});
const responseTime = (Date.now() - startTime) / 1000;
this.recordMetrics(proxyUrl, responseTime, true);
return response;
} catch (error) {
this.recordMetrics(proxyUrl, 0, false);
throw error;
} finally {
this.connections.set(proxyUrl, this.connections.get(proxyUrl) - 1);
}
}
parseProxyUrl(proxyUrl) {
const url = new URL(proxyUrl);
return {
host: url.hostname,
port: parseInt(url.port)
};
}
recordMetrics(proxyUrl, responseTime, success) {
const m = this.metrics.get(proxyUrl);
m.total++;
if (success) {
m.success++;
m.responseTimes.push(responseTime);
if (m.responseTimes.length > 1000) {
m.responseTimes.shift();
}
} else {
m.failed++;
}
}
getStats() {
const stats = [];
this.proxies.forEach(proxy => {
const m = this.metrics.get(proxy.url);
const avgResponseTime = m.responseTimes.length > 0
? m.responseTimes.reduce((a, b) => a + b, 0) / m.responseTimes.length
: 0;
const successRate = m.total > 0 ? (m.success / m.total * 100) : 0;
stats.push({
proxy: proxy.url,
totalRequests: m.total,
successRate: successRate.toFixed(2),
avgResponseTime: avgResponseTime.toFixed(3),
activeConnections: this.connections.get(proxy.url)
});
});
return stats;
}
}
// Utilizzo
const proxies = [
{ url: 'http://proxy1.example.com:8080', weight: 5 },
{ url: 'http://proxy2.example.com:8080', weight: 5 }
];
const balancer = new ProxyLoadBalancer(proxies, 'round_robin', 60);
async function parseProducts() {
const urls = Array.from({ length: 1000 }, (_, i) =>
`https://www.wildberries.ru/catalog/${i}/detail.aspx`
);
for (const url of urls) {
try {
const response = await balancer.makeRequest(url);
console.log(`Success: ${url}`);
} catch (error) {
console.error(`Error: ${url} - ${error.message}`);
}
}
console.log('\n=== Statistiche ===');
console.log(balancer.getStats());
}
parseProducts();Ottimizzazione per compiti specifici: parsing, API, automazione
Compiti diversi richiedono strategie diverse di bilanciamento del carico. Per il parsing di marketplace (Wildberries, Ozon, Avito) è ottimale una strategia con rotazione per numero di richieste e distribuzione geografica. Utilizza proxy residenziali di diverse città della Russia, cambia proxy ogni 50-100 richieste, aggiungi ritardi casuali tra le richieste (1-3 secondi) per simulare il comportamento umano.
Per lavorare con API di social media (Instagram, Facebook, VK) è fondamentale la stabilità dell'indirizzo IP all'interno di una singola sessione. Utilizza l'algoritmo IP Hash o sticky sessions: tutte le richieste di un account devono passare attraverso lo stesso proxy. Questo previene cambiamenti sospetti nella geolocalizzazione, che possono causare il blocco dell'account. Si raccomandano proxy mobili, poiché i social media si fidano di più degli IP mobili rispetto a quelli residenziali o dei data center.
# Esempio di sticky sessions per Instagram
class StickySessionBalancer:
def __init__(self, proxies):
self.proxies = proxies
self.session_map = {} # {account_id: proxy_url}
self.proxy_usage = defaultdict(int)
def get_proxy_for_account(self, account_id):
"""Restituisce un proxy fisso per l'account"""
if account_id in self.session_map:
return self.session_map[account_id]
# Selezioniamo il proxy meno utilizzato
proxy = min(self.proxies, key=lambda p: self.proxy_usage[p])
self.session_map[account_id] = proxy
self.proxy_usage[proxy] += 1
return proxy
def release_account(self, account_id):
"""Libera il proxy al termine del lavoro con l'account"""
if account_id in self.session_map:
proxy = self.session_map[account_id]
self.proxy_usage[proxy] -= 1
del self.session_map[account_id]