توازن الحمل عبر البروكسي: كيفية توسيع عملية التحليل وطلبات API إلى أكثر من 10000 طلب في الثانية

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

المادة موجهة للمطورين والمتخصصين الفنيين الذين يعملون مع زحف البيانات، أتمتة طلبات API أو يديرون بركسيات كبيرة لمهام الأعمال.

لماذا تحتاج إلى توازن الحمل عبر البروكسي

يحقق توازن الحمل عبر البروكسي عدة مشاكل حرجة في الأنظمة ذات الحمل العالي. المشكلة الأولى والأهم هي الحماية من الحظر. عندما ترسل آلاف الطلبات إلى مورد واحد (سوق، API شبكة اجتماعية، محرك بحث)، يرى الخادم المستهدف نشاطًا غير طبيعي مرتفعًا من عنوان IP واحد ويقوم بحظره. توزيع الطلبات بين عشرات أو مئات البروكسيات يجعل نشاطك يبدو كأفعال مستخدمين عاديين.

المشكلة الثانية هي الأداء. يحتوي خادم البروكسي الواحد على عرض نطاق محدود (عادةً 100-1000 ميجابت في الثانية) ويمكنه معالجة عدد محدود من الاتصالات المتزامنة. عند الزحف على 10000 صفحة في الدقيقة أو إرسال طلبات API جماعية، سيصبح البروكسي واحدًا نقطة اختناق في النظام. يسمح التوازن بتوسيع عرض النطاق أفقيًا عن طريق إضافة بروكسيات جديدة إلى البركة.

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

السبب الرابع هو التوزيع الجغرافي. تعرض العديد من الخدمات محتوى أو أسعار مختلفة اعتمادًا على منطقة المستخدم. يسمح التوازن بين البروكسيات من دول ومدن مختلفة بجمع البيانات من جميع المناطق المستهدفة في وقت واحد. على سبيل المثال، لمراقبة أسعار Ozon، تحتاج إلى بروكسيات من موسكو، سانت بطرسبرغ، يكاترينبرغ وغيرها من المدن الكبرى.

هيكلية نظام توزيع الحمل

تتكون الهيكلية الكلاسيكية لنظام توازن الحمل عبر البروكسي من عدة مكونات. في المستوى الأعلى يوجد موازن الحمل (load balancer) — وحدة برمجية تستقبل المهام الواردة (طلبات الزحف، استدعاءات API) وتوزعها بين البروكسيات المتاحة. يمكن أن يعمل موازن الحمل كخدمة منفصلة أو يكون مدمجًا في التطبيق.

المكون الثاني هو مدير بركة البروكسي (proxy pool manager). يحتفظ بقائمة بجميع البروكسيات المتاحة مع خصائصها: عنوان IP، المنفذ، البروتوكول (HTTP/SOCKS5)، الموقع الجغرافي، النوع (سكني، موبايل، مركز بيانات)، الحالة الحالية (نشط، محظور، قيد الفحص). يتولى مدير البركة مسؤولية إضافة بروكسيات جديدة، إزالة غير العاملة وإجراء فحص دوري للتوافر.

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

المكون الرابع هو محدد معدل الطلبات (rate limiter). يتأكد من أن كل بروكسي لا يتجاوز معدل الطلبات المسموح به إلى المورد المستهدف. على سبيل المثال، إذا كنت تقوم بزحف Instagram وتعلم أن المنصة تحظر IP بعد 60 طلبًا في الدقيقة، فلن يسمح محدد معدل الطلبات بإرسال أكثر من 50 طلبًا عبر بروكسي واحد في الدقيقة، مما يترك هامش أمان.

العنصر الخامس هو نظام القياسات والتحليلات. يجمع بيانات عن أداء كل بروكسي: عدد الطلبات الناجحة، عدد الأخطاء، متوسط زمن الاستجابة، نسبة الحظر. تُستخدم هذه البيانات لتحسين خوارزميات التوازن واكتشاف البروكسيات المشكلة.

خوارزميات التوازن: Round Robin، أقل الاتصالات، Weighted

خوارزمية Round Robin هي أبسط وأكثر طرق التوازن شيوعًا. تقوم بالتناوب على البروكسيات من القائمة: الطلب الأول يمر عبر البروكسي رقم 1، الثاني عبر البروكسي رقم 2، الثالث عبر البروكسي رقم 3، وهكذا. عندما تنتهي القائمة، يعود موازن الحمل إلى البداية. توزع هذه الخوارزمية الحمل بشكل متساوٍ إذا كانت جميع البروكسيات لها نفس الأداء.

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

# الاستخدام
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"طلب {i+1} → {proxy}")

خوارزمية أقل الاتصالات (Least Connections) تختار البروكسي الذي لديه أقل عدد من الاتصالات النشطة في الوقت الحالي. هذا مفيد عندما تكون الطلبات لها مدة تنفيذ مختلفة. على سبيل المثال، عند الزحف، قد يتم معالجة طلب واحد في 100 مللي ثانية، بينما قد يستغرق الآخر 5 ثوانٍ (إذا كانت الصفحة بطيئة في التحميل). تقوم خوارزمية أقل الاتصالات تلقائيًا بتوجيه الطلبات الجديدة إلى البروكسيات الأقل تحميلًا.

class LeastConnectionsBalancer:
    def __init__(self, proxies):
        self.proxies = {proxy: 0 for proxy in proxies}
    
    def get_next_proxy(self):
        # اختيار البروكسي مع أقل عدد من الاتصالات
        proxy = min(self.proxies.items(), key=lambda x: x[1])[0]
        self.proxies[proxy] += 1
        return proxy
    
    def release_proxy(self, proxy):
        # تقليل العداد عند انتهاء الطلب
        self.proxies[proxy] -= 1

# الاستخدام مع مدير السياق
balancer = LeastConnectionsBalancer(proxies)

def make_request(url):
    proxy = balancer.get_next_proxy()
    try:
        # تنفيذ الطلب عبر البروكسي المختار
        response = requests.get(url, proxies={"http": proxy})
        return response
    finally:
        balancer.release_proxy(proxy)

Weighted Round Robin (التدوير الموزون) يوسع Round Robin التقليدية، حيث يتم تعيين وزن لكل بروكسي بناءً على أدائه أو جودته. تحصل البروكسيات ذات الوزن الأكبر على المزيد من الطلبات. هذا مفيد عندما يكون لديك بروكسيات ذات جودة مختلفة: على سبيل المثال، بروكسيات سكنية متميزة بسرعة عالية وبروكسيات مراكز بيانات رخيصة مع قيود.

class WeightedRoundRobinBalancer:
    def __init__(self, weighted_proxies):
        # weighted_proxies = [(proxy, weight), ...]
        self.proxies = []
        for proxy, weight in weighted_proxies:
            # إضافة البروكسي إلى القائمة بعدد الوزن
            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

# الاستخدام مع الأوزان
weighted_proxies = [
    ("http://premium-proxy1.com:8080", 5),  # جودة عالية
    ("http://premium-proxy2.com:8080", 5),
    ("http://cheap-proxy1.com:8080", 2),    # جودة منخفضة
    ("http://cheap-proxy2.com:8080", 1)     # جودة منخفضة جدًا
]

balancer = WeightedRoundRobinBalancer(weighted_proxies)

خوارزمية Random (الاختيار العشوائي) تختار البروكسي بشكل عشوائي من القائمة المتاحة. إنها أسهل في التنفيذ من Round Robin وتعمل بشكل جيد عند وجود عدد كبير من البروكسيات (100+). يتم توزيع الطلبات بشكل عشوائي تلقائيًا عند وجود حجم كافٍ من الطلبات. العيب هو إمكانية وجود فترات قصيرة من الحمل غير المتوازن.

IP Hash — خوارزمية تختار البروكسي بناءً على هاش URL المستهدف أو معلمة أخرى. يضمن ذلك أن الطلبات إلى نفس المورد دائمًا تمر عبر نفس البروكسي. مفيد للمواقع التي تستخدم الجلسات أو تتطلب تسلسل الطلبات من نفس IP (مثل، المصادقة + الحصول على البيانات).

إدارة بركسيات البروكسي: التدوير وفحوصات الصحة

تتطلب إدارة فعالة لبركة البروكسي مراقبة مستمرة لحالة كل بروكسي وتدوير تلقائي. فحص الصحة هو تحقق دوري من توافر البروكسي عن طريق إرسال طلب اختبار. عادةً ما يتم استخدام طلب GET بسيط إلى خدمة موثوقة (مثل httpbin.org أو نقطة النهاية الخاصة بك)، والتي تعيد معلومات عن عنوان 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):
        """يتحقق من عمل البروكسي"""
        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": "صحي",
                    "response_time": response_time,
                    "timestamp": datetime.now(),
                    "ip": response.json().get("origin")
                }
            else:
                return {
                    "status": "غير صحي",
                    "error": f"HTTP {response.status_code}",
                    "timestamp": datetime.now()
                }
        except requests.exceptions.Timeout:
            return {
                "status": "غير صحي",
                "error": "انتهاء الوقت",
                "timestamp": datetime.now()
            }
        except Exception as e:
            return {
                "status": "غير صحي",
                "error": str(e),
                "timestamp": datetime.now()
            }

# الاستخدام
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)")

يجب أن يقوم نظام إدارة البركة تلقائيًا بتمييز البروكسيات غير العاملة على أنها غير متاحة وإعادة فحصها دوريًا. استراتيجية شائعة هي نمط قاطع الدائرة: بعد ثلاث فحوصات غير ناجحة على التوالي، يتم استبعاد البروكسي من البركة لمدة 5-10 دقائق، ثم يتم فحصه مرة أخرى. إذا كانت الفحص ناجحًا، يتم إعادة البروكسي إلى البركة النشطة.

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):
        """يضيف بروكسي إلى البركة"""
        self.proxies[proxy_url] = {
            "url": proxy_url,
            "status": "نشط",
            "failures": 0,
            "last_check": None,
            "cooldown_until": None,
            "metadata": metadata or {}
        }
    
    def get_active_proxies(self):
        """يعود بقائمة البروكسيات النشطة"""
        now = datetime.now()
        active = []
        
        for proxy_url, info in self.proxies.items():
            # تحقق مما إذا كان البروكسي في فترة التبريد
            if info["cooldown_until"] and now < info["cooldown_until"]:
                continue
            
            if info["status"] == "نشط":
                active.append(proxy_url)
        
        return active
    
    def mark_failure(self, proxy_url):
        """يسجل محاولة غير ناجحة لاستخدام البروكسي"""
        if proxy_url not in self.proxies:
            return
        
        info = self.proxies[proxy_url]
        info["failures"] += 1
        
        if info["failures"] >= self.max_failures:
            # الانتقال إلى فترة التبريد
            info["status"] = "فترة تبريد"
            info["cooldown_until"] = datetime.now() + timedelta(seconds=self.cooldown_seconds)
            print(f"تم نقل البروكسي {proxy_url} إلى فترة التبريد حتى {info['cooldown_until']}")
    
    def mark_success(self, proxy_url):
        """يسجل استخدام ناجح للبروكسي"""
        if proxy_url not in self.proxies:
            return
        
        info = self.proxies[proxy_url]
        info["failures"] = 0
        info["status"] = "نشط"
        info["cooldown_until"] = None

تدوير البروكسي هو استراتيجية لتغيير البروكسي تلقائيًا عبر فترات زمنية معينة أو بعد عدد معين من الطلبات. هناك عدة طرق: تدوير زمني (تغيير كل 5-10 دقائق)، تدوير حسب عدد الطلبات (تغيير بعد 100-500 طلب)، تدوير حسب الجلسات (بروكسي واحد لجلسة زحف واحدة). يعتمد اختيار الاستراتيجية على متطلبات الموقع المستهدف.

تحديد معدل الطلبات ومراقبة تكرار الطلبات

تحديد معدل الطلبات هو مكون حيوي في نظام توازن الحمل. يمنع تجاوز معدل الطلبات المسموح به إلى المورد المستهدف، مما قد يؤدي إلى حظر البروكسي. هناك خوارزميتان رئيسيتان: دلو الرموز (Token Bucket) ودلو مثقوب (Leaky Bucket).

يعمل دلو الرموز على النحو التالي: كل بروكسي لديه "دلو" افتراضي يحتوي على رموز. كل رمز يمنح الحق في طلب واحد. يتم ملء الدلو تدريجيًا بالرموز بمعدل محدد (على سبيل المثال، 10 رموز في الثانية). السعة القصوى للدلو محدودة (على سبيل المثال، 50 رمزًا). عندما يأتي طلب، يتحقق النظام من وجود الرموز: إذا كانت موجودة — يأخذ رمزًا واحدًا وينفذ الطلب، إذا لم تكن موجودة — ينتظر ظهور رمز جديد.

import time
from threading import Lock

class TokenBucketRateLimiter:
    def __init__(self, rate, capacity):
        """
        rate: عدد الرموز في الثانية
        capacity: الحد الأقصى لعدد الرموز في الدلو
        """
        self.rate = rate
        self.capacity = capacity
        self.tokens = capacity
        self.last_update = time.time()
        self.lock = Lock()
    
    def _add_tokens(self):
        """يضيف الرموز بناءً على الوقت المنقضي"""
        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):
        """يحاول الحصول على عدد محدد من الرموز"""
        with self.lock:
            self._add_tokens()
            
            if self.tokens >= tokens:
                self.tokens -= tokens
                return True
            return False
    
    def wait_and_acquire(self, tokens=1):
        """ينتظر ظهور الرموز ويحصل عليها"""
        while not self.acquire(tokens):
            # حساب وقت الانتظار
            wait_time = (tokens - self.tokens) / self.rate
            time.sleep(wait_time)

# الاستخدام لكل بروكسي
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()  # انتظر توفر الرمز
    
    response = requests.get(url, proxies={"http": proxy_url})
    return response

يعمل دلو مثقوب (Leaky Bucket) بطريقة مختلفة: يتم وضع الطلبات في قائمة (دلو) تتسرب "بسرعة ثابتة". إذا امتلأ الدلو، يتم رفض الطلبات الجديدة أو وضعها في الانتظار. تضمن هذه الخوارزمية توزيع الحمل بشكل أكثر توازنًا عبر الزمن، لكنها قد تؤدي إلى تأخيرات خلال فترات النشاط المفاجئ.

تتطلب المنصات المستهدفة المختلفة حدودًا مختلفة. يسمح API Instagram بحوالي 200 طلب في الساعة لكل IP (حوالي 3.3 طلب في الدقيقة). تحظر Wildberries بعد 100-200 طلب في الدقيقة من IP واحد. يسمح Google Search بـ 10-20 طلب في الدقيقة. يجب أن تأخذ نظام تحديد معدل الطلبات في الاعتبار هذه القيود وإضافة هامش أمان (عادةً 20-30%).

مراقبة الأداء والتوسع التلقائي

تتطلب نظام توازن الحمل الفعال مراقبة مستمرة للقياسات الرئيسية. المجموعة الأولى من القياسات هي أداء البروكسي: متوسط زمن الاستجابة (response time)، نسبة الطلبات الناجحة (success rate)، عدد الانقطاعات، عدد الأخطاء 4xx و 5xx. تسمح هذه البيانات باكتشاف البروكسيات المشكلة وتحسين خوارزميات التوازن.

class ProxyMetrics:
    def __init__(self):
        self.metrics = {}  # {proxy_url: metrics_dict}
    
    def record_request(self, proxy_url, response_time, status_code, success):
        """يسجل قياسات الطلب"""
        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):
        """يعيد نسبة الطلبات الناجحة"""
        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):
        """يعيد متوسط زمن الاستجابة"""
        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):
        """يعيد تقرير كامل عن البروكسي"""
        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)
        }

المجموعة الثانية من القياسات هي الأداء العام للنظام: إجمالي عدد الطلبات في الثانية (RPS — requests per second)، متوسط التأخير (latency)، حجم قائمة الطلبات، عدد البروكسيات النشطة، نسبة استخدام بركة البروكسي. تظهر هذه القياسات ما إذا كان النظام يتعامل مع الحمل أو يحتاج إلى توسيع.

يسمح التوسع التلقائي (auto-scaling) بإضافة أو إزالة البروكسيات ديناميكيًا بناءً على الحمل. استراتيجية بسيطة: إذا تجاوز متوسط تحميل البركة 80% لمدة 5 دقائق، يقوم النظام تلقائيًا بإضافة بروكسيات جديدة. إذا انخفض الحمل إلى أقل من 30% لمدة 15 دقيقة، يقوم النظام بإزالة البروكسيات الزائدة لتوفير الموارد.

لاستخدام القياسات بصريًا، استخدم Grafana مع Prometheus أو ELK Stack. أنشئ لوحات معلومات مع الرسوم البيانية: RPS على مر الزمن، توزيع زمن الاستجابة، أفضل 10 بروكسيات الأسرع/الأبطأ، خريطة الأخطاء حسب الأنواع. سيساعد ذلك في الكشف السريع عن المشاكل وتحسين النظام.

تنفيذ عملي على Python و Node.js

دعونا نلقي نظرة على تنفيذ كامل لنظام توازن الحمل على Python باستخدام المكتبات الشائعة. يجمع هذا المثال جميع المكونات الموصوفة: موازن الحمل، مدير البركة، فحوصات الصحة، تحديد معدل الطلبات والمراقبة.

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: قائمة البروكسيات [{"url": "...", "weight": 1}, ...]
        algorithm: round_robin، least_connections، weighted، random
        rate_limit: الحد الأقصى لعدد الطلبات في الثانية لكل بروكسي
        """
        self.proxies = proxies
        self.algorithm = algorithm
        self.rate_limit = rate_limit
        
        # حالة موازن الحمل
        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
        })
        
        # تهيئة محددات المعدل
        for proxy in proxies:
            proxy_url = proxy["url"]
            self.rate_limiters[proxy_url] = TokenBucketRateLimiter(
                rate=rate_limit,
                capacity=rate_limit * 5
            )
        
        self.lock = Lock()
        
        # بدء فحص الصحة في الخلفية
        self.health_check_thread = Thread(target=self._health_check_loop, daemon=True)
        self.health_check_thread.start()
    
    def get_next_proxy(self):
        """يختار البروكسي التالي وفقًا للخوارزمية"""
        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):
        """ينفذ الطلب عبر موازن الحمل"""
        proxy_url = self.get_next_proxy()
        
        # انتظر توفر معدل الطلب
        self.rate_limiters[proxy_url].wait_and_acquire()
        
        # زيادة عداد الاتصالات
        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
            
            # سجل القياسات
            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:
            # تقليل عداد الاتصالات
            with self.lock:
                self.connections[proxy_url] -= 1
    
    def _record_metrics(self, proxy_url, response_time, success, status_code):
        """يسجل قياسات الطلب"""
        with self.lock:
            m = self.metrics[proxy_url]
            m["total"] += 1
            if success:
                m["success"] += 1
                m["response_times"].append(response_time)
                # احتفظ فقط بأحدث 1000 قيمة
                if len(m["response_times"]) > 1000:
                    m["response_times"].pop(0)
            else:
                m["failed"] += 1
    
    def _health_check_loop(self):
        """فحص الصحة في الخلفية للبروكسيات"""
        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"] = "صحي" if response.status_code == 200 else "غير صحي"
                except:
                    with self.lock:
                        proxy["status"] = "غير صحي"
            
            time.sleep(60)  # تحقق كل دقيقة
    
    def get_stats(self):
        """يعيد الإحصائيات لجميع البروكسيات"""
        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", "غير معروف"),
                    "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 stats

استخدام هذا الموازن لزحف سوق:

# إعداد الموازن
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 طلبًا في الدقيقة لكل بروكسي
)

# زحف قائمة المنتجات
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)
        # معالجة الاستجابة
        results.append({"url": url, "status": "success", "data": response.text})
    except Exception as e:
        results.append({"url": url, "status": "error", "error": str(e)})
    
    # كل 100 طلب نعرض الإحصائيات
    if len(results) % 100 == 0:
        stats = balancer.get_stats()
        for stat in stats:
            print(f"{stat['proxy']}: {stat['success_rate']}% نجاح، "
                  f"{stat['avg_response_time']}s متوسط الاستجابة")

# الإحصائيات النهائية
print("\n=== الإحصائيات النهائية ===")
for stat in balancer.get_stats():
    print(f"{stat['proxy']}:")
    print(f"  إجمالي الطلبات: {stat['total_requests']}")
    print(f"  نسبة النجاح: {stat['success_rate']}%")
    print(f"  متوسط زمن الاستجابة: {stat['avg_response_time']}s")
    print(f"  الحالة: {stat['status']}")

بالنسبة لـ Node.js، يمكنك استخدام هيكل مشابه مع مكتبات axios لطلبات HTTP و node-rate-limiter للتحكم في المعدل:

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();
    
    // تهيئة محددات المعدل
    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);
    
    // انتظر توفر معدل الطلب
    await limiter.removeTokens(1);
    
    // زيادة عداد الاتصالات
    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;
  }
}

// الاستخدام
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(`نجاح: ${url}`);
    } catch (error) {
      console.error(`خطأ: ${url} - ${error.message}`);
    }
  }
  
  console.log('\n=== الإحصائيات ===');
  console.log(balancer.getStats());
}

parseProducts();

تحسين لمهام محددة: الزحف، API، الأتمتة

تتطلب المهام المختلفة استراتيجيات مختلفة لتوازن الحمل. بالنسبة لزحف الأسواق (Wildberries، Ozon، Avito)، فإن الاستراتيجية المثلى هي التدوير حسب عدد الطلبات والتوزيع الجغرافي. استخدم بروكسيات سكنية من مدن مختلفة في روسيا، قم بتغيير البروكسيات كل 50-100 طلبًا، وأضف تأخيرات عشوائية بين الطلبات (1-3 ثوانٍ) لمحاكاة سلوك الإنسان.

بالنسبة للعمل مع API الشبكات الاجتماعية (Instagram، Facebook، VK)، فإن استقرار عنوان IP خلال جلسة واحدة يكون حاسمًا. استخدم خوارزمية IP Hash أو الجلسات الثابتة: يجب أن تمر جميع الطلبات لحساب واحد عبر نفس البروكسي. يمنع ذلك تغيير الموقع الجغرافي المريب، مما قد يؤدي إلى حظر الحساب. يُوصى باستخدام بروكسيات موبايل، حيث تثق الشبكات الاجتماعية في IPs الموبايل أكثر من السكنية أو مراكز البيانات.

# مثال على الجلسات الثابتة لـ 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):
        """يعيد البروكسي الثابت للحساب"""
        if account_id in self.session_map:
            return self.session_map[account_id]
        
        # اختيار البروكسي الأقل تحميلًا
        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):
        """يحرر البروكسي عند الانتهاء من العمل مع الحساب"""
        if account_id in self.session_map:
            proxy = self.session_map[account_id]
            self.proxy_usage[proxy] -= 1
            del self.session_map[account_id]

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