استراتيجيات تدوير البروكسيات للاستخراج واسع النطاق

لِمَ التناوب المحترف هو أمر أساسي لـ (لارج سكالي)

عندما تتقدم من المئات إلى الملايين من الطلبات، فإن أي بي سي واحد يصبح مسؤولية. وتتتبع المواقع الشبكية أنماط الطلب لكل مجموعة من شركاء التنفيذ، وستتعرض أو تحظر العناوين التي تتجاوز السلوك العادي للتصفير. التناوب يوزع طلباتك عبر العديد من شركاء التنفيذ حتى لا يتراكم عنوان واحد نشاطاً كافياً لكشفه

والفرق بين نهج التناوب السذاجة واستراتيجية حسنة التصميم يمكن أن يعني الفرق بين معدل النجاح البالغ 95 في المائة ونسبة 40 في المائة. ويشمل هذا الدليل استراتيجيات التناوب الرئيسية الأربعة، عند استخدام كل منها، وكيفية تنفيذها بأمثلة على مدونة العمل.

هذه المادة جزء من دليل كامل للدعاوى الإلكترونية مجموعة. ابدأ من هناك إذا كنت بحاجة إلى مفاهيم المحترفين

الاستراتيجية 1: التناوب عند الطلب

النهج الأبسط: كل طلب يحصل على آي بي جديدوهذا مثالي لجمع البيانات عديمة الجنسية حيثما يكون كل طلب مستقلاً - النظر في الأسعار، والاستفسارات في نظام تخطيط الموارد في المؤسسة، وفتيات المنتجات.

متى تستخدم

• فهرس كبير حيث يكون كل من الجبهة الثورية المستقلة
• SERP monitoring across many keywords
• أي مهمة لا تتطلب الكعك أو الدورة

Python Implementation

import requests
PROXY = "http://USERNAME:PASSWORD@gate.proxyhat.com:8080"
def fetch_with_rotation(urls: list[str]) -> list[str]:
    """Each request automatically gets a fresh IP via the rotating gateway."""
    results = []
    session = requests.Session()
    session.proxies = {"http": PROXY, "https": PROXY}
    for url in urls:
        try:
            resp = session.get(url, timeout=30)
            resp.raise_for_status()
            results.append(resp.text)
        except requests.RequestException as e:
            print(f"Failed {url}: {e}")
            results.append(None)
    return results
# Each request through gate.proxyhat.com uses a different IP
pages = fetch_with_rotation([
    "https://example.com/product/1",
    "https://example.com/product/2",
    "https://example.com/product/3",
])

Node.js Implementation

const HttpsProxyAgent = require('https-proxy-agent');
const fetch = require('node-fetch');
const agent = new HttpsProxyAgent('http://USERNAME:PASSWORD@gate.proxyhat.com:8080');
async function fetchWithRotation(urls) {
  const results = [];
  for (const url of urls) {
    try {
      const res = await fetch(url, { agent, timeout: 30000 });
      results.push(await res.text());
    } catch (err) {
      console.error(`Failed ${url}: ${err.message}`);
      results.push(null);
    }
  }
  return results;
}

التنفيذ

package main
import (
    "fmt"
    "io"
    "net/http"
    "net/url"
    "time"
)
func fetchWithRotation(urls []string) []string {
    proxyURL, _ := url.Parse("http://USERNAME:PASSWORD@gate.proxyhat.com:8080")
    client := &http.Client{
        Transport: &http.Transport{Proxy: http.ProxyURL(proxyURL)},
        Timeout:   30 * time.Second,
    }
    results := make([]string, len(urls))
    for i, u := range urls {
        resp, err := client.Get(u)
        if err != nil {
            fmt.Printf("Failed %s: %v\n", u, err)
            continue
        }
        body, _ := io.ReadAll(resp.Body)
        resp.Body.Close()
        results[i] = string(body)
    }
    return results
}

الاستراتيجية 2: التناوب الزمني (دورات السماء)

وتحتاج بعض مهام الخردة إلى نفس البرنامج من أجل مجموعة من الطلبات ذات الصلة - أي وضع قائمة مبتورة، أو تسمية عملية تفتيش متعددة الخطوات، أو الإبقاء على دورة تسجيلية. التناوب الزمني (أو الجلسات اللزجة) تبقي نفس البرنامج المخصص لفترة محددة، عادة 1-30 دقيقة.

متى تستخدم

• الزحف المدفوع (الصفحة 1، 2، 3... من النتائج)
• المهام التي تتطلب البسكويت أو استمرار الدورة
• تبسيط الأنماط الواقعية للتصفح

خطة التنفيذ

مع (بروكسي هات) الجلسات اللزجة يتم التحكم بها عبر بارامتر الدورة في وثائق تفويضك وتحتفظ كل دورة فريدة من نوعها بالهوية نفسها لفترة محددة:

import requests
import uuid
def create_sticky_session(duration_label: str = "10m"):
    """Create a session that maintains the same IP."""
    session_id = uuid.uuid4().hex[:8]
    proxy = f"http://USERNAME-session-{session_id}:PASSWORD@gate.proxyhat.com:8080"
    session = requests.Session()
    session.proxies = {"http": proxy, "https": proxy}
    return session
# All requests through this session use the same IP
session = create_sticky_session()
page1 = session.get("https://example.com/listings?page=1")
page2 = session.get("https://example.com/listings?page=2")
page3 = session.get("https://example.com/listings?page=3")

Node.js Sticky Session

const HttpsProxyAgent = require('https-proxy-agent');
const fetch = require('node-fetch');
const crypto = require('crypto');
function createStickyAgent() {
  const sessionId = crypto.randomBytes(4).toString('hex');
  return new HttpsProxyAgent(
    `http://USERNAME-session-${sessionId}:PASSWORD@gate.proxyhat.com:8080`
  );
}
async function crawlPaginated(baseUrl, pages) {
  const agent = createStickyAgent(); // Same IP for all pages
  const results = [];
  for (let page = 1; page <= pages; page++) {
    const res = await fetch(`${baseUrl}?page=${page}`, { agent });
    results.push(await res.text());
  }
  return results;
}

الاستراتيجية 3: التناوب على أساس الفشل

بدلاً من التناوب على كل طلب أو على جهاز توقيت التناوب على أساس الفشل يَستمرُّ بإستعمال آي بي حتى يُغلق، ثمّ يَتغيّرُ. This maximizes the value of each IP by using it as long as it works.

متى تستخدم

• الأهداف التي لا يمكن التنبؤ بها
• الخردة والوعي في الميزانية حيث تريد الطلبات القصوى
• زحف طويل المدى حيث بعض شركاء التنفيذ الساعات الماضية ودقائق أخرى

Implementation with Automatic Failover

import requests
import uuid
from time import sleep
class FailureBasedRotator:
    """Rotates proxy only when the current IP fails."""
    BLOCK_SIGNALS = [403, 429, 503]
    MAX_RETRIES = 3
    def __init__(self):
        self.session_id = None
        self.requests_on_current_ip = 0
        self._new_session()
    def _new_session(self):
        self.session_id = uuid.uuid4().hex[:8]
        self.requests_on_current_ip = 0
        proxy = f"http://USERNAME-session-{self.session_id}:PASSWORD@gate.proxyhat.com:8080"
        self.session = requests.Session()
        self.session.proxies = {"http": proxy, "https": proxy}
    def fetch(self, url: str) -> str | None:
        for attempt in range(self.MAX_RETRIES):
            try:
                resp = self.session.get(url, timeout=30)
                if resp.status_code in self.BLOCK_SIGNALS:
                    print(f"Blocked (HTTP {resp.status_code}) after "
                          f"{self.requests_on_current_ip} requests. Rotating...")
                    self._new_session()
                    sleep(1)
                    continue
                resp.raise_for_status()
                self.requests_on_current_ip += 1
                return resp.text
            except requests.RequestException:
                self._new_session()
                sleep(1)
        return None
# Usage
rotator = FailureBasedRotator()
for url in urls:
    html = rotator.fetch(url)

Go Implementation with Failover

package main
import (
    "crypto/rand"
    "encoding/hex"
    "fmt"
    "io"
    "net/http"
    "net/url"
    "time"
)
type FailureRotator struct {
    client    *http.Client
    sessionID string
    reqCount  int
}
func NewFailureRotator() *FailureRotator {
    r := &FailureRotator{}
    r.rotate()
    return r
}
func (r *FailureRotator) rotate() {
    b := make([]byte, 4)
    rand.Read(b)
    r.sessionID = hex.EncodeToString(b)
    r.reqCount = 0
    proxyStr := fmt.Sprintf("http://USERNAME-session-%s:PASSWORD@gate.proxyhat.com:8080", r.sessionID)
    proxyURL, _ := url.Parse(proxyStr)
    r.client = &http.Client{
        Transport: &http.Transport{Proxy: http.ProxyURL(proxyURL)},
        Timeout:   30 * time.Second,
    }
}
func (r *FailureRotator) Fetch(target string) (string, error) {
    for attempt := 0; attempt < 3; attempt++ {
        resp, err := r.client.Get(target)
        if err != nil {
            r.rotate()
            time.Sleep(time.Second)
            continue
        }
        defer resp.Body.Close()
        if resp.StatusCode == 403 || resp.StatusCode == 429 || resp.StatusCode == 503 {
            fmt.Printf("Blocked after %d requests. Rotating...\n", r.reqCount)
            r.rotate()
            time.Sleep(time.Second)
            continue
        }
        body, _ := io.ReadAll(resp.Body)
        r.reqCount++
        return string(body), nil
    }
    return "", fmt.Errorf("all retries exhausted for %s", target)
}

الاستراتيجية 4: التناوب الجغرافي

وعندما تخريد المحتوى المحلي - نتائج البحث والتسعير والتوافر - تحتاج شركاء التنفيذ من مواقع جغرافية محددة. التناوب الجغرافي يخصص شركاء التنفيذ من البلدان أو المدن المستهدفة للحصول على بيانات محلية دقيقة.

متى تستخدم

• SERP scraping للتفتيش المحلي
• رصد الأسعار عبر المناطق
• التحقق من مدى توافر المحتوى (المحتوى المقيد بالأرقام الجغرافية)
• التحقق في أسواق محددة

التنفيذ مع الأهداف القطرية

import requests
from concurrent.futures import ThreadPoolExecutor
COUNTRIES = ["us", "gb", "de", "fr", "jp"]
def fetch_localized(url: str, country: str) -> dict:
    """Fetch URL through a proxy in the specified country."""
    proxy = f"http://USERNAME-country-{country}:PASSWORD@gate.proxyhat.com:8080"
    try:
        resp = requests.get(url, proxies={"http": proxy, "https": proxy}, timeout=30)
        return {"country": country, "status": resp.status_code, "body": resp.text}
    except requests.RequestException as e:
        return {"country": country, "status": 0, "error": str(e)}
def scrape_all_regions(url: str) -> list[dict]:
    """Fetch the same URL from multiple countries in parallel."""
    with ThreadPoolExecutor(max_workers=len(COUNTRIES)) as executor:
        futures = [executor.submit(fetch_localized, url, c) for c in COUNTRIES]
        return [f.result() for f in futures]
# Get localized pricing from 5 countries simultaneously
results = scrape_all_regions("https://example.com/product/pricing")
for r in results:
    print(f"{r['country'].upper()}: HTTP {r['status']}")

See available targeting options on the أماكن الإقامة الصفحة

الاستراتيجيات المشتركة: النهج الهجين

In practice, large-scale scraping projects combine multiple strategies. هنا نمط يستخدم التناوب في كل طلب للاكتشاف، وجلسات ملتصقة للزحف العميق، والفشل

import requests
import uuid
from enum import Enum
class RotationMode(Enum):
    PER_REQUEST = "per_request"
    STICKY = "sticky"
    FAILURE_BASED = "failure_based"
class HybridRotator:
    def __init__(self, mode: RotationMode = RotationMode.PER_REQUEST):
        self.mode = mode
        self.session_id = None
        self.failure_count = 0
        self._init_session()
    def _init_session(self):
        if self.mode == RotationMode.PER_REQUEST:
            proxy = "http://USERNAME:PASSWORD@gate.proxyhat.com:8080"
        else:
            self.session_id = self.session_id or uuid.uuid4().hex[:8]
            proxy = f"http://USERNAME-session-{self.session_id}:PASSWORD@gate.proxyhat.com:8080"
        self.session = requests.Session()
        self.session.proxies = {"http": proxy, "https": proxy}
    def force_rotate(self):
        """Force a new IP regardless of mode."""
        self.session_id = uuid.uuid4().hex[:8]
        self.failure_count = 0
        self._init_session()
    def fetch(self, url: str) -> str | None:
        try:
            resp = self.session.get(url, timeout=30)
            if resp.status_code in [403, 429, 503]:
                self.failure_count += 1
                if self.failure_count >= 2:
                    self.force_rotate()
                return None
            self.failure_count = 0
            return resp.text
        except requests.RequestException:
            self.failure_count += 1
            if self.failure_count >= 2:
                self.force_rotate()
            return None
# Discovery phase: rotate every request
discovery = HybridRotator(RotationMode.PER_REQUEST)
sitemap_urls = [discovery.fetch(url) for url in seed_urls]
# Deep crawl phase: sticky sessions per site section
crawler = HybridRotator(RotationMode.STICKY)
for section_url in section_urls:
    pages = [crawler.fetch(f"{section_url}?page={i}") for i in range(1, 11)]
    crawler.force_rotate()  # New IP for next section

مقارنة استراتيجية التناوب

أفضل الممارسات للتناوب في سكال

• احترم الروبوتات التناوب لا يعفيك من كونك مواطناً صالحاً تحقق من القواعد و تكريم توجيهات التسلل
• إضافة تأخيرات واقعية. حتى مع التناوب، تفجر مئات الطلبات في الثانية تبدو روبوتية. مضافا إليها حالات التأخير العشوائي الثانية بين الطلبات
• رصد معدلات النجاح. تعقّب رموز مركز HTTP لكل موقع مستهدف. انخفاض أقل من 90 في المائة يعني تناوبك يحتاج للضغط
• مركب مع تناوب الرأس تناوب شركاء التنفيذ وحده ليس كافياً سلاسل المستعملين والمرشدين الآخرين لتجنب الكشف عن بصمات الأصابع.
• استخدام التراجع على الفشل. عندما يُغلق أي بي، انتظر قبل إعادة التجريب. وتمنع التخلف العرضي )١ و ٢ و ٤ و ٨( من تبديد الطلبات على أهداف معادية مؤقتا.

لفهم كم عدد شركاء التنفيذ الذين تحتاجهم لدعم إستراتيجيتك للتناوب كم عدد المحترفين الذين تحتاجون للتشويش؟- للاطلاع على لمحة عامة شاملة عن هيكل الخردة، زيارة لنا دليل كامل للدعاوى الإلكترونية.

مستعدة لتنفيذ هذه الاستراتيجيات؟ تحقق من Python SDK.. Node SDKأو Go SDK من أجل الدمج الجاهز للإنتاج، أو استكشاف خطط التسعير للبدء

الأسئلة المتكررة

ما هي أفضل استراتيجية للتناوب على الإنترنت؟

والتناوب فيما يتعلق بالطلب هو أسلم تقصير لمعظم مهام الخردة. ويضمن أن يستخدم كل طلب تعريفاً مختلفاً يجعل كشف النمط أصعب بكثير. أما بالنسبة للمهام التي تتطلب استمرار الدورة (الاختراع، وتدفقات قطع الأشجار)، فتستخدم بدلاً من ذلك جلسات لاصقة.

ما مدى سرعتي في تناوب العملاء؟

بالنسبة للتناوب في كل طلب يحصل تلقائياً على آي بي للجلساتِ اللزجةِ، 5-10 دقائقِ a عجز جيد. وتتوقف المدة المثلى على الهدف - وقد تتطلب المواقع العدوانية عقد دورات أقصر )١-٢ دقيقة(، في حين تتساهل المواقع العصيبة مع ٣٠ دقيقة.

هل يمكنني الجمع بين استراتيجيات مختلفة للتناوب؟

أجل، وعليكِ أن تقومي بمشاريع معقدة Use per-request rotation for discovery and URL collection, sticky sessions for deep crawling, and failure-based circulation as a fallback when IPs get blocked. ويبين النهج الهجين في هذا الدليل كيف.

هل يقوم (بروكسي هات) بالتناوب تلقائياً؟

نعم ويتلقى كل طلب عبر بوابة بروكسيهات (G.proxyhat.com:8080) تلقائياً مجموعة مختلفة من شركاء التنفيذ من المجمع السكني. للجلساتِ اللزجةِ، يَضْفُ a بارامتر الدورةِ إلى وثائقِ تفويضِكَ. ولا حاجة إلى أي إدارة يدوية لقائمة شركاء التنفيذ.