نحوه رفع خطاهای timeout هنگام کار از طریق پروکسی

درخواست متوقف شد، اسکریپت با خطای TimeoutError شکست خورد، داده‌ها دریافت نشدند. وضعیت آشنایی است؟ خطاهای timeout از طریق پروکسی — یکی از متداول‌ترین مشکلات در parsing و خودکارسازی است. دلایل را بررسی کنیم و راه‌حل‌های مشخصی ارائه دهیم.

چرا خطاهای timeout رخ می‌دهند

Timeout — نه یک مشکل، بلکه علامت یک مشکل است. قبل از درمان، باید دلیل را درک کنیم:

سرور پروکسی کند. سرور بارگذاری‌شده یا پروکسی از نظر جغرافیایی دور، تأخیر را به هر درخواست اضافه می‌کند. اگر timeout شما 10 ثانیه باشد و پروکسی در 12 ثانیه پاسخ دهد — خطا رخ می‌دهد.

مسدود شدن در سمت وب‌سایت هدف. وب‌سایت می‌تواند درخواست‌های مشکوک را عمداً «معلق» کند به جای رد صریح. این تاکتیک علیه ربات‌ها است — اتصال را بی‌نهایت باز نگه دارید.

مشکلات DNS. پروکسی باید دامنه را حل کند. اگر سرور DNS پروکسی کند یا در دسترس نباشد — درخواست در مرحله اتصال معلق می‌ماند.

تنظیم نادرست timeout‌ها. یک timeout کلی برای همه — خطای متداول است. Connect timeout و read timeout — چیزهای متفاوتی هستند و باید جداگانه تنظیم شوند.

مشکلات شبکه. از دست دادن بسته‌ها، اتصال ناپایدار پروکسی، مشکلات مسیریابی — همه اینها منجر به timeout می‌شوند.

انواع timeout‌ها و تنظیم آن‌ها

اکثر کتابخانه‌های HTTP از چندین نوع timeout پشتیبانی می‌کنند. درک تفاوت بین آن‌ها — کلید تنظیم صحیح است.

Connect timeout

زمان برای برقراری اتصال TCP با پروکسی و سرور هدف. اگر پروکسی در دسترس نباشد یا سرور پاسخ ندهد — این timeout فعال می‌شود. مقدار توصیه‌شده: 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,      # Timeout کل
        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 باید با هوشمندی انجام شود.

تأخیر نمایی

درخواست‌های مجدد را بدون توقف به سرور نفرستید. از exponential 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()

چرخش پروکسی هنگام timeout‌ها

اگر یک پروکسی دائماً timeout می‌دهد — مشکل در آن است. راه‌حل منطقی: به پروکسی دیگری بروید.

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"Timeout از طریق {proxy}، پروکسی دیگری را امتحان می‌کنیم...")
            continue
    
    raise Exception(f"بعد از {max_attempts} تلاش نتوانستیم داده‌ها را دریافت کنیم")

هنگام استفاده از پروکسی‌های مسکونی با چرخش خودکار این منطق ساده‌تر می‌شود — فراهم‌کننده خود IP را در هر درخواست یا بر اساس بازه‌ی تعیین‌شده تغییر می‌دهد.

درخواست‌های ناهمزمان با کنترل timeout‌ها

در parsing انبوه درخواست‌های همزمان ناکارآمد هستند. رویکرد ناهمزمان امکان پردازش صدها URL به‌طور موازی را می‌دهد، اما نیاز به کار احتیاط‌آمیز با 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]:
    """بارگذاری یک URL با مدیریت timeout"""
    
    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‌ها و موازی‌سازی"""
    
    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}, Timeout‌ها: {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"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("مستقیم:", measure_request(url))
print("از طریق پروکسی:", measure_request(url, proxy))

مرحله 3: انواع مختلف پروکسی را بررسی کنید

Timeout‌ها می‌توانند به نوع پروکسی بستگی داشته باشند:

مرحله 4: جزئیات را ثبت کنید

import logging
import requests
from requests.adapters import HTTPAdapter

# فعال کردن debug-logging
logging.basicConfig(level=logging.DEBUG)
logging.getLogger("urllib3").setLevel(logging.DEBUG)

# اکنون تمام مراحل درخواست را خواهید دید:
# - حل کردن DNS
# - برقراری اتصال
# - ارسال درخواست
# - دریافت پاسخ

چک‌لیست حل خطاهای timeout

الگوریتم کوتاه اقدام هنگام timeout‌ها:

1. نوع timeout را مشخص کنید — connect یا read؟ اینها مشکلات متفاوتی هستند.
2. پروکسی را جداگانه بررسی کنید — آیا اصلاً کار می‌کند؟ تأخیر چقدر است؟
3. Timeout‌ها را افزایش دهید — ممکن است مقادیر برای نوع پروکسی شما بیش از حد تهاجمی باشند.
4. Retry با backoff اضافه کنید — timeout‌های منفرد طبیعی هستند، مهم پایداری است.
5. چرخش را تنظیم کنید — خودکار به پروکسی دیگری بروید هنگام مشکلات.
6. موازی‌سازی را محدود کنید — درخواست‌های همزمان بیش از حد پروکسی را بارگذاری می‌کنند.
7. وب‌سایت هدف را بررسی کنید — ممکن است درخواست‌های شما را مسدود یا throttle کند.

نتیجه‌گیری

خطاهای timeout از طریق پروکسی — مشکل قابل حل است. در اکثر موارد تنظیم صحیح timeout‌ها برای نوع پروکسی، اضافه کردن منطق retry و پیاده‌سازی چرخش هنگام شکست کافی است. برای کارهایی با نیازهای بالا برای پایداری از پروکسی‌های مسکونی با چرخش خودکار استفاده کنید — جزئیات بیشتر در proxycove.com.