Come correggere gli errori di timeout quando si lavora tramite proxy
La richiesta si è bloccata, lo script è caduto con un errore TimeoutError, i dati non sono stati ricevuti. Una situazione familiare? Gli errori di timeout tramite proxy sono uno dei problemi più comuni durante il parsing e l'automazione. Analizziamo le cause e forniamo soluzioni concrete.
Perché si verificano gli errori di timeout
Il timeout non è un unico problema, ma un sintomo. Prima di curare, è necessario comprendere la causa:
Server proxy lento. Un server sovraccarico o un proxy geograficamente distante aggiungono ritardo a ogni richiesta. Se il vostro timeout è di 10 secondi, ma il proxy risponde in 12 secondi, si verifica un errore.
Blocco dal lato del sito di destinazione. Il sito può intenzionalmente "sospendere" le richieste sospette invece di rifiutarle esplicitamente. Questa è una tattica contro i bot: mantenere la connessione aperta indefinitamente.
Problemi con DNS. Il proxy deve risolvere il dominio. Se il server DNS del proxy è lento o non disponibile, la richiesta si blocca nella fase di connessione.
Configurazione errata dei timeout. Un singolo timeout generale per tutto è un errore comune. Il timeout di connessione e il timeout di lettura sono cose diverse e devono essere configurati separatamente.
Problemi di rete. Perdita di pacchetti, connessione instabile del proxy, problemi di routing: tutto questo porta a timeout.
Tipi di timeout e loro configurazione
La maggior parte delle librerie HTTP supporta diversi tipi di timeout. Comprendere la differenza tra loro è la chiave per una corretta configurazione.
Timeout di connessione
Tempo per stabilire una connessione TCP con il proxy e il server di destinazione. Se il proxy non è disponibile o il server non risponde, questo timeout si attiva. Valore consigliato: 5-10 secondi.
Timeout di lettura
Tempo di attesa dei dati dopo aver stabilito la connessione. Il server si è connesso, ma rimane silenzioso: il timeout di lettura si attiva. Per le pagine normali: 15-30 secondi. Per API pesanti: 60+ secondi.
Timeout totale
Tempo totale per l'intera richiesta dall'inizio alla fine. Protezione contro le connessioni bloccate. Di solito: connessione + lettura + margine.
Esempio di configurazione in Python con la libreria requests:
import requests
proxies = {
"http": "http://user:pass@proxy.example.com:8080",
"https": "http://user:pass@proxy.example.com:8080"
}
# Tupla: (connect_timeout, read_timeout)
timeout = (10, 30)
try:
response = requests.get(
"https://target-site.com/api/data",
proxies=proxies,
timeout=timeout
)
except requests.exceptions.ConnectTimeout:
print("Impossibile connettersi al proxy o al server")
except requests.exceptions.ReadTimeout:
print("Il server non ha inviato i dati in tempo")
Per aiohttp (Python asincrono):
import aiohttp
import asyncio
async def fetch_with_timeout():
timeout = aiohttp.ClientTimeout(
total=60, # Timeout totale
connect=10, # Per la connessione
sock_read=30 # Per la lettura dei dati
)
async with aiohttp.ClientSession(timeout=timeout) as session:
async with session.get(
"https://target-site.com/api/data",
proxy="http://user:pass@proxy.example.com:8080"
) as response:
return await response.text()
Logica di retry: l'approccio corretto
Il timeout non è sempre un errore fatale. Spesso una richiesta ripetuta ha successo. Ma il retry deve essere fatto con intelligenza.
Ritardo esponenziale
Non bombardate il server con richieste ripetute senza pausa. Utilizzate il backoff esponenziale: ogni tentativo successivo ha un ritardo crescente.
import requests
import time
import random
def fetch_with_retry(url, proxies, max_retries=3):
"""Richiesta con retry e ritardo esponenziale"""
for attempt in range(max_retries):
try:
response = requests.get(
url,
proxies=proxies,
timeout=(10, 30)
)
response.raise_for_status()
return response
except (requests.exceptions.Timeout,
requests.exceptions.ConnectionError) as e:
if attempt == max_retries - 1:
raise # Ultimo tentativo: lanciamo l'errore
# Ritardo esponenziale: 1s, 2s, 4s...
# + jitter casuale per evitare onde di richieste
delay = (2 ** attempt) + random.uniform(0, 1)
print(f"Tentativo {attempt + 1} non riuscito: {e}")
print(f"Nuovo tentativo tra {delay:.1f} secondi...")
time.sleep(delay)
Libreria tenacity
Per il codice di produzione è più conveniente utilizzare soluzioni pronte:
from tenacity import retry, stop_after_attempt, wait_exponential
from tenacity import retry_if_exception_type
import requests
@retry(
stop=stop_after_attempt(3),
wait=wait_exponential(multiplier=1, min=1, max=10),
retry=retry_if_exception_type((
requests.exceptions.Timeout,
requests.exceptions.ConnectionError
))
)
def fetch_data(url, proxies):
response = requests.get(url, proxies=proxies, timeout=(10, 30))
response.raise_for_status()
return response.json()
Rotazione del proxy in caso di timeout
Se un proxy continua a dare timeout, il problema è in esso. La soluzione logica: passare a un altro.
import requests
from collections import deque
from dataclasses import dataclass, field
from typing import Optional
import time
@dataclass
class ProxyManager:
"""Gestore proxy con tracciamento dei tentativi falliti"""
proxies: list
max_failures: int = 3
cooldown_seconds: int = 300
_failures: dict = field(default_factory=dict)
_cooldown_until: dict = field(default_factory=dict)
def get_proxy(self) -> Optional[str]:
"""Ottenere un proxy funzionante"""
current_time = time.time()
for proxy in self.proxies:
# Saltiamo i proxy in cooldown
if self._cooldown_until.get(proxy, 0) > current_time:
continue
return proxy
return None # Tutti i proxy in cooldown
def report_failure(self, proxy: str):
"""Segnalare una richiesta non riuscita"""
self._failures[proxy] = self._failures.get(proxy, 0) + 1
if self._failures[proxy] >= self.max_failures:
# Mettiamo il proxy in cooldown
self._cooldown_until[proxy] = time.time() + self.cooldown_seconds
self._failures[proxy] = 0
print(f"Proxy {proxy} messo in cooldown")
def report_success(self, proxy: str):
"""Resettare il contatore degli errori al successo"""
self._failures[proxy] = 0
def fetch_with_rotation(url, proxy_manager, max_attempts=5):
"""Richiesta con cambio automatico del proxy in caso di errori"""
for attempt in range(max_attempts):
proxy = proxy_manager.get_proxy()
if not proxy:
raise Exception("Nessun proxy disponibile")
proxies = {"http": proxy, "https": proxy}
try:
response = requests.get(url, proxies=proxies, timeout=(10, 30))
response.raise_for_status()
proxy_manager.report_success(proxy)
return response
except (requests.exceptions.Timeout,
requests.exceptions.ConnectionError):
proxy_manager.report_failure(proxy)
print(f"Timeout tramite {proxy}, proviamo un altro...")
continue
raise Exception(f"Impossibile ottenere i dati dopo {max_attempts} tentativi")
Quando si utilizzano proxy residenziali con rotazione automatica, questa logica si semplifica: il provider stesso commuta l'IP ad ogni richiesta o secondo l'intervallo specificato.
Richieste asincrone con controllo dei timeout
Nel parsing di massa, le richieste sincrone sono inefficienti. L'approccio asincrono consente di elaborare centinaia di URL in parallelo, ma richiede un'attenta gestione dei timeout.
import aiohttp
import asyncio
from typing import List, Tuple
async def fetch_one(
session: aiohttp.ClientSession,
url: str,
semaphore: asyncio.Semaphore
) -> Tuple[str, str | None, str | None]:
"""Caricamento di un singolo URL con gestione del timeout"""
async with semaphore: # Limitiamo il parallelismo
try:
async with session.get(url) as response:
content = await response.text()
return (url, content, None)
except asyncio.TimeoutError:
return (url, None, "timeout")
except aiohttp.ClientError as e:
return (url, None, str(e))
async def fetch_all(
urls: List[str],
proxy: str,
max_concurrent: int = 10
) -> List[Tuple[str, str | None, str | None]]:
"""Caricamento di massa con controllo dei timeout e del parallelismo"""
timeout = aiohttp.ClientTimeout(total=45, connect=10, sock_read=30)
semaphore = asyncio.Semaphore(max_concurrent)
connector = aiohttp.TCPConnector(
limit=max_concurrent,
limit_per_host=5 # Non più di 5 connessioni per host
)
async with aiohttp.ClientSession(
timeout=timeout,
connector=connector
) as session:
# Impostiamo il proxy per tutte le richieste
tasks = [
fetch_one(session, url, semaphore)
for url in urls
]
results = await asyncio.gather(*tasks)
# Statistiche
success = sum(1 for _, content, _ in results if content)
timeouts = sum(1 for _, _, error in results if error == "timeout")
print(f"Riusciti: {success}, Timeout: {timeouts}")
return results
# Utilizzo
async def main():
urls = [f"https://example.com/page/{i}" for i in range(100)]
results = await fetch_all(
urls,
proxy="http://user:pass@proxy.example.com:8080",
max_concurrent=10
)
asyncio.run(main())
Importante: Non impostate un parallelismo troppo elevato. 50-100 richieste simultanee attraverso un singolo proxy sono già molte. Meglio 10-20 con più proxy.
Diagnostica: come trovare la causa
Prima di modificare le impostazioni, determinate la fonte del problema.
Passaggio 1: Verificare il proxy direttamente
# Semplice test tramite curl con misurazione del tempo
curl -x http://user:pass@proxy:8080 \
-w "Connect: %{time_connect}s\nTotal: %{time_total}s\n" \
-o /dev/null -s \
https://httpbin.org/get
Se time_connect è maggiore di 5 secondi, il problema è con il proxy o la rete verso di esso.
Passaggio 2: Confrontare con una richiesta diretta
import requests
import time
def measure_request(url, proxies=None):
start = time.time()
try:
r = requests.get(url, proxies=proxies, timeout=30)
elapsed = time.time() - start
return f"OK: {elapsed:.2f}s, status: {r.status_code}"
except Exception as e:
elapsed = time.time() - start
return f"FAIL: {elapsed:.2f}s, error: {type(e).__name__}"
url = "https://target-site.com"
proxy = {"http": "http://proxy:8080", "https": "http://proxy:8080"}
print("Direttamente:", measure_request(url))
print("Tramite proxy:", measure_request(url, proxy))
Passaggio 3: Verificare diversi tipi di proxy
I timeout possono dipendere dal tipo di proxy:
| Tipo di proxy | Ritardo tipico | Timeout consigliato |
|---|---|---|
| Datacenter | 50-200 ms | Connessione: 5s, Lettura: 15s |
| Residenziale | 200-800 ms | Connessione: 10s, Lettura: 30s |
| Mobile | 300-1500 ms | Connessione: 15s, Lettura: 45s |
Passaggio 4: Registrare i dettagli
import logging
import requests
from requests.adapters import HTTPAdapter
# Abilitiamo il debug logging
logging.basicConfig(level=logging.DEBUG)
logging.getLogger("urllib3").setLevel(logging.DEBUG)
# Ora vedrete tutte le fasi della richiesta:
# - Risoluzione DNS
# - Stabilimento della connessione
# - Invio della richiesta
# - Ricezione della risposta
Checklist per la risoluzione degli errori di timeout
Algoritmo breve per agire quando si verificano timeout:
- Identificare il tipo di timeout — connessione o lettura? Sono problemi diversi.
- Verificare il proxy separatamente — funziona? Quale ritardo ha?
- Aumentare i timeout — forse i valori sono troppo aggressivi per il vostro tipo di proxy.
- Aggiungere retry con backoff — i timeout singoli sono normali, l'importante è la resilienza.
- Configurare la rotazione — passare automaticamente a un altro proxy in caso di problemi.
- Limitare il parallelismo — troppe richieste simultanee sovraccaricano il proxy.
- Verificare il sito di destinazione — potrebbe bloccare o throttle le vostre richieste.
Conclusione
Gli errori di timeout tramite proxy sono un problema risolvibile. Nella maggior parte dei casi è sufficiente configurare correttamente i timeout in base al tipo di proxy, aggiungere logica di retry e implementare la rotazione in caso di guasti. Per compiti con elevati requisiti di stabilità, utilizzate proxy residenziali con rotazione automatica: ulteriori informazioni su proxycove.com.