प्रॉक्सी के माध्यम से काम करते समय timeout त्रुटियों को कैसे ठीक करें

अनुरोध हैंग हो गया, स्क्रिप्ट TimeoutError के साथ गिर गई, डेटा प्राप्त नहीं हुआ। परिचित स्थिति? प्रॉक्सी के माध्यम से Timeout-त्रुटियां पार्सिंग और ऑटोमेशन में सबसे आम समस्याओं में से एक हैं। आइए कारणों को समझें और ठोस समाधान दें।

timeout त्रुटियां क्यों होती हैं

Timeout एक समस्या नहीं है, बल्कि एक लक्षण है। इलाज से पहले, कारण समझना जरूरी है:

धीमा प्रॉक्सी-सर्वर। ओवरलोडेड सर्वर या भौगोलिक रूप से दूरस्थ प्रॉक्सी प्रत्येक अनुरोध में देरी जोड़ता है। यदि आपका टाइमआउट 10 सेकंड है, लेकिन प्रॉक्सी 12 सेकंड में जवाब देता है — त्रुटि।

लक्ष्य साइट पर ब्लॉकिंग। साइट स्पष्ट अस्वीकार के बजाय संदिग्ध अनुरोधों को जानबूझकर "लटका" सकती है। यह बॉट्स के खिलाफ एक रणनीति है — कनेक्शन को अनिश्चित काल के लिए खुला रखना।

DNS समस्याएं। प्रॉक्सी को डोमेन को रिज़ॉल्व करना चाहिए। यदि प्रॉक्सी का DNS-सर्वर धीमा या अनुपलब्ध है — अनुरोध कनेक्शन चरण पर हैंग हो जाता है।

गलत टाइमआउट कॉन्फ़िगरेशन। सब कुछ के लिए एक सामान्य टाइमआउट — एक सामान्य गलती। Connect timeout और read timeout — अलग चीजें हैं, और उन्हें अलग से कॉन्फ़िगर करना चाहिए।

नेटवर्क समस्याएं। पैकेट हानि, प्रॉक्सी का अस्थिर कनेक्शन, रूटिंग समस्याएं — यह सब टाइमआउट की ओर ले जाता है।

टाइमआउट के प्रकार और उनका कॉन्फ़िगरेशन

अधिकांश HTTP-लाइब्रेरी कई प्रकार के टाइमआउट का समर्थन करती हैं। उनके बीच अंतर को समझना सही कॉन्फ़िगरेशन की कुंजी है।

Connect timeout

प्रॉक्सी और लक्ष्य सर्वर के साथ TCP-कनेक्शन स्थापित करने का समय। यदि प्रॉक्सी अनुपलब्ध है या सर्वर प्रतिक्रिया नहीं करता है — यह टाइमआउट काम करेगा। अनुशंसित मान: 5-10 सेकंड।

Read timeout

कनेक्शन स्थापित होने के बाद डेटा की प्रतीक्षा का समय। सर्वर जुड़ा है, लेकिन चुप है — read timeout काम करेगा। सामान्य पृष्ठों के लिए: 15-30 सेकंड। भारी API के लिए: 60+ सेकंड।

Total timeout

शुरुआत से अंत तक पूरे अनुरोध पर कुल समय। हैंग किए गए कनेक्शन से सुरक्षा। आमतौर पर: connect + read + मार्जिन।

Python में requests लाइब्रेरी के साथ कॉन्फ़िगरेशन का उदाहरण:

import requests

proxies = {
    "http": "http://user:pass@proxy.example.com:8080",
    "https": "http://user:pass@proxy.example.com:8080"
}

# टपल: (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("प्रॉक्सी या सर्वर से कनेक्ट नहीं हो सके")
except requests.exceptions.ReadTimeout:
    print("सर्वर ने समय पर डेटा नहीं भेजा")

aiohttp के लिए (अतुल्यकालिक Python):

import aiohttp
import asyncio

async def fetch_with_timeout():
    timeout = aiohttp.ClientTimeout(
        total=60,      # कुल टाइमआउट
        connect=10,    # कनेक्शन पर
        sock_read=30   # डेटा पढ़ने पर
    )
    
    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()

Retry-लॉजिक: सही दृष्टिकोण

Timeout हमेशा घातक त्रुटि नहीं है। अक्सर दोहराया गया अनुरोध सफलतापूर्वक पास हो जाता है। लेकिन retry को बुद्धिमानी से करना चाहिए।

घातीय विलंब

बिना विराम के दोहराए गए अनुरोधों से सर्वर को न मारें। घातीय backoff का उपयोग करें: प्रत्येक अगला प्रयास — बढ़ते हुए विलंब के साथ।

import requests
import time
import random

def fetch_with_retry(url, proxies, max_retries=3):
    """retry और घातीय विलंब के साथ अनुरोध"""
    
    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  # अंतिम प्रयास — त्रुटि फेंकें
            
            # घातीय विलंब: 1s, 2s, 4s...
            # + यादृच्छिक jitter अनुरोधों की लहरें न बनाने के लिए
            delay = (2 ** attempt) + random.uniform(0, 1)
            print(f"प्रयास {attempt + 1} विफल: {e}")
            print(f"{delay:.1f} सेकंड में दोहराएं...")
            time.sleep(delay)

tenacity लाइब्रेरी

Production-कोड के लिए तैयार समाधान का उपयोग करना अधिक सुविधाजनक है:

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()

टाइमआउट पर प्रॉक्सी रोटेशन

यदि एक प्रॉक्सी लगातार टाइमआउट देता है — समस्या इसमें है। तार्किक समाधान: दूसरे पर स्विच करें।

import requests
from collections import deque
from dataclasses import dataclass, field
from typing import Optional
import time

@dataclass
class ProxyManager:
    """विफल प्रयासों को ट्रैक करने वाला प्रॉक्सी प्रबंधक"""
    
    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]:
        """काम करने वाला प्रॉक्सी प्राप्त करें"""
        current_time = time.time()
        
        for proxy in self.proxies:
            # cooldown पर प्रॉक्सी को छोड़ें
            if self._cooldown_until.get(proxy, 0) > current_time:
                continue
            return proxy
        
        return None  # सभी प्रॉक्सी cooldown पर हैं
    
    def report_failure(self, proxy: str):
        """विफल अनुरोध की रिपोर्ट करें"""
        self._failures[proxy] = self._failures.get(proxy, 0) + 1
        
        if self._failures[proxy] >= self.max_failures:
            # प्रॉक्सी को cooldown पर भेजें
            self._cooldown_until[proxy] = time.time() + self.cooldown_seconds
            self._failures[proxy] = 0
            print(f"प्रॉक्सी {proxy} को cooldown पर भेजा गया")
    
    def report_success(self, proxy: str):
        """सफलता पर त्रुटि काउंटर रीसेट करें"""
        self._failures[proxy] = 0


def fetch_with_rotation(url, proxy_manager, max_attempts=5):
    """त्रुटियों पर स्वचालित प्रॉक्सी स्विच के साथ अनुरोध"""
    
    for attempt in range(max_attempts):
        proxy = proxy_manager.get_proxy()
        
        if not proxy:
            raise Exception("कोई उपलब्ध प्रॉक्सी नहीं")
        
        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"{proxy} के माध्यम से टाइमआउट, दूसरा आजमाते हैं...")
            continue
    
    raise Exception(f"{max_attempts} प्रयासों के बाद डेटा प्राप्त नहीं कर सके")

आवासीय प्रॉक्सी का उपयोग करते समय स्वचालित रोटेशन के साथ यह लॉजिक सरल हो जाता है — प्रदाता हर अनुरोध पर या निर्दिष्ट अंतराल पर IP को स्वयं स्विच करता है।

टाइमआउट नियंत्रण के साथ अतुल्यकालिक अनुरोध

बड़े पैमाने पर पार्सिंग में सिंक्रोनस अनुरोध अक्षम हैं। अतुल्यकालिक दृष्टिकोण सैकड़ों URL को समानांतर में संसाधित करने की अनुमति देता है, लेकिन टाइमआउट के साथ सावधान काम की आवश्यकता है।

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]:
    """टाइमआउट हैंडलिंग के साथ एक URL लोड करें"""
    
    async with semaphore:  # समानांतरता को सीमित करें
        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]]:
    """टाइमआउट और समानांतरता नियंत्रण के साथ बल्क लोडिंग"""
    
    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  # एक होस्ट पर 5 से अधिक कनेक्शन नहीं
    )
    
    async with aiohttp.ClientSession(
        timeout=timeout,
        connector=connector
    ) as session:
        # सभी अनुरोधों के लिए प्रॉक्सी सेट करें
        tasks = [
            fetch_one(session, url, semaphore) 
            for url in urls
        ]
        results = await asyncio.gather(*tasks)
    
    # सांख्यिकी
    success = sum(1 for _, content, _ in results if content)
    timeouts = sum(1 for _, _, error in results if error == "timeout")
    print(f"सफल: {success}, टाइमआउट: {timeouts}")
    
    return results


# उपयोग
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())

महत्वपूर्ण: बहुत अधिक समानांतरता न सेट करें। एक प्रॉक्सी के माध्यम से 50-100 एक साथ अनुरोध — पहले से ही बहुत है। कई प्रॉक्सी के साथ 10-20 बेहतर है।

निदान: कारण कैसे खोजें

सेटिंग्स बदलने से पहले, समस्या का स्रोत निर्धारित करें।

चरण 1: प्रॉक्सी को सीधे जांचें

# समय माप के साथ curl के माध्यम से सरल परीक्षण
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

यदि time_connect 5 सेकंड से अधिक है — प्रॉक्सी या इसके नेटवर्क में समस्या है।

चरण 2: सीधे अनुरोध से तुलना करें

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"ठीक है: {elapsed:.2f}s, स्थिति: {r.status_code}"
    except Exception as e:
        elapsed = time.time() - start
        return f"विफल: {elapsed:.2f}s, त्रुटि: {type(e).__name__}"

url = "https://target-site.com"
proxy = {"http": "http://proxy:8080", "https": "http://proxy:8080"}

print("सीधे:", measure_request(url))
print("प्रॉक्सी के माध्यम से:", measure_request(url, proxy))

चरण 3: विभिन्न प्रकार के प्रॉक्सी जांचें

टाइमआउट प्रॉक्सी के प्रकार पर निर्भर हो सकते हैं:

चरण 4: विवरण लॉग करें

import logging
import requests
from requests.adapters import HTTPAdapter

# debug-लॉगिंग सक्षम करें
logging.basicConfig(level=logging.DEBUG)
logging.getLogger("urllib3").setLevel(logging.DEBUG)

# अब आप अनुरोध के सभी चरण देखेंगे:
# - DNS रिज़ॉल्विंग
# - कनेक्शन स्थापना
# - अनुरोध भेजना
# - प्रतिक्रिया प्राप्त करना

timeout-त्रुटियों को हल करने की चेकलिस्ट

टाइमआउट होने पर कार्यों का संक्षिप्त एल्गोरिदम:

1. टाइमआउट का प्रकार निर्धारित करें — connect या read? ये अलग समस्याएं हैं।
2. प्रॉक्सी को अलग से जांचें — क्या यह काम करता है? विलंब क्या है?
3. टाइमआउट बढ़ाएं — संभवतः मान आपके प्रॉक्सी प्रकार के लिए बहुत आक्रामक हैं।
4. backoff के साथ retry जोड़ें — एकल टाइमआउट सामान्य हैं, महत्वपूर्ण है स्थिरता।
5. रोटेशन कॉन्फ़िगर करें — समस्याओं पर स्वचालित रूप से दूसरे प्रॉक्सी पर स्विच करें।
6. समानांतरता सीमित करें — बहुत सारे एक साथ अनुरोध प्रॉक्सी को ओवरलोड करते हैं।
7. लक्ष्य साइट जांचें — संभवतः यह आपके अनुरोधों को ब्लॉक या throttle कर रहा है।

निष्कर्ष

प्रॉक्सी के माध्यम से Timeout-त्रुटियां एक हल करने योग्य समस्या हैं। अधिकांश मामलों में, प्रॉक्सी प्रकार के लिए टाइमआउट को सही तरीके से कॉन्फ़िगर करना, retry-लॉजिक जोड़ना और विफलताओं पर रोटेशन लागू करना पर्याप्त है। उच्च स्थिरता आवश्यकताओं वाले कार्यों के लिए, स्वचालित रोटेशन के साथ आवासीय प्रॉक्सी का उपयोग करें — proxycove.com पर अधिक जानकारी।