التوسع التلقائي لعمّال حل CAPTCHA

في CaptchaAI تُحاسَب على عدد الخيوط المتزامنة لا على عدد الحلول، لذا فإن ضبط عدد العمّال ديناميكيًا هو ما يحدّد تكلفتك وإنتاجيتك في آنٍ واحد. التوسع التلقائي يربط عدد العمّال بعمق قائمة الانتظار والرصيد ومعدل الحل، فترفع الطاقة عند الذروة وتُطلق الخيوط عند الهدوء دون تدخّل يدوي.

المجموعة الثابتة من العمّال تخسر في الاتجاهين: خيوط خاملة تدفع ثمنها بلا عائد وقت الهدوء، واختناق يُطيل زمن الانتظار وقت الضغط. الهدف من هذا الدليل أن يصبح عدد العمّال دالة في الطلب الفعلي، مضبوطة بحدٍّ أعلى لا يتجاوز خيوط خطتك.

متى يصبح التوسع التلقائي ضروريًا

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

• منصّة تجارة إلكترونية خليجية تُطلق تخفيضات مفاجئة في مواسم مثل الجمعة البيضاء أو رمضان، فيقفز حجم عمليات إتمام الشراء المحمية بـ CAPTCHA أضعافًا خلال دقائق.
• بوابة حجز تذاكر أو مواعيد حكومية تفتح دفعة محدودة في توقيت معلن، فترتفع طلبات التحقق دفعة واحدة ثم تهدأ.
• عملية استخراج بيانات مجدولة ليلًا لتخفيف الضغط، تتطلّب طاقة عالية لساعات قليلة ثم لا شيء.

في هذه الحالات، عدد العمّال المناسب للذروة يكون مُكلفًا ومُهدِرًا بقية اليوم، والعدد المناسب للهدوء يخنق الذروة. التوسع التلقائي يحلّ التوتر بينهما.

إشارات ضبط عدد العمّال

قرار التوسع لا يُبنى على مؤشّر واحد بل على مجموعة إشارات تُقرأ معًا. الجدول التالي يلخّص متى ترفع العدد ومتى تخفضه:

القاعدة العملية: اجعل رفع العدد سريع الاستجابة وخفضه بطيئًا ومتحفّظًا، حتى لا يتذبذب النظام بين حالتي التوسع والتقليص كلما تغيّر الحمل لحظيًا.

واقرأ الإشارات على طبقتين:

• إشارات محفِّزة: عمق قائمة الانتظار ونسبة الانشغال، وهما ما يقرّر الرفع فعليًا.
• إشارات حارسة: الرصيد ومعدل الخطأ، يمنعان الرفع حين لا يجدي.

التوسع التلقائي على مستوى الخيوط

بما أنّ استدعاء CaptchaAI عملية مقيّدة بالإدخال/الإخراج (I/O-bound)، فإنّ توسيع الخيوط داخل عملية واحدة هو الأسلوب الأخفّ والأنسب للأغلبية. المجمّع التالي يبدأ بحدٍّ أدنى من الخيوط، ثم يقرأ عمق قائمة الانتظار ونسبة الانشغال كل عشر ثوانٍ ليضيف أو يُطلق خيوطًا:

import os
import time
import threading
import requests
import json
import redis


class AutoScalingPool:
    """Dynamically scale CaptchaAI worker threads."""

    def __init__(self, api_key, redis_url="redis://localhost:6379"):
        self.api_key = api_key
        self.redis = redis.from_url(redis_url)
        self.base = "https://ocr.captchaai.com"
        self.queue_key = "captcha:tasks"
        self.results_key = "captcha:results"

        self.min_workers = 2
        self.max_workers = 20
        self.workers = []
        self.active_count = 0
        self.lock = threading.Lock()
        self.running = True

    def start(self):
        """Start the pool with minimum workers."""
        for _ in range(self.min_workers):
            self._add_worker()

        # Start scaler in background
        scaler = threading.Thread(target=self._scaling_loop, daemon=True)
        scaler.start()
        print(f"Pool started with {self.min_workers} workers")

    def _add_worker(self):
        """Add a worker thread."""
        if len(self.workers) >= self.max_workers:
            return
        t = threading.Thread(target=self._worker_loop, daemon=True)
        t.start()
        self.workers.append(t)

    def _remove_worker(self):
        """Signal one worker to stop (lazy removal)."""
        if len(self.workers) <= self.min_workers:
            return
        self.workers.pop()  # Thread will exit on next idle cycle

    def _worker_loop(self):
        """Worker loop: fetch and process tasks."""
        while self.running and threading.current_thread() in self.workers:
            result = self.redis.blpop(self.queue_key, timeout=10)
            if result is None:
                continue

            _, raw = result
            task = json.loads(raw)
            task_id = task["id"]

            with self.lock:
                self.active_count += 1

            try:
                token = self._solve(task["method"], task["params"])
                self.redis.hset(self.results_key, task_id, json.dumps({
                    "status": "success", "token": token,
                }))
            except Exception as e:
                self.redis.hset(self.results_key, task_id, json.dumps({
                    "status": "error", "error": str(e),
                }))
            finally:
                with self.lock:
                    self.active_count -= 1

    def _scaling_loop(self):
        """Periodically adjust worker count."""
        while self.running:
            time.sleep(10)

            queue_depth = self.redis.llen(self.queue_key)
            current = len(self.workers)
            utilization = (
                self.active_count / current * 100 if current > 0 else 0
            )

            # Scale up: queue growing and workers busy
            if queue_depth > 20 and utilization > 70:
                new_count = min(current + 2, self.max_workers)
                while len(self.workers) < new_count:
                    self._add_worker()
                print(f"Scaled up: {current} → {len(self.workers)} workers")

            # Scale down: queue empty and workers idle
            elif queue_depth < 5 and utilization < 20:
                target = max(current - 1, self.min_workers)
                while len(self.workers) > target:
                    self._remove_worker()
                if len(self.workers) < current:
                    print(f"Scaled down: {current} → {len(self.workers)} workers")

    def _solve(self, method, params, timeout=120):
        data = {"key": self.api_key, "method": method, "json": 1}
        data.update(params)

        resp = requests.post(
            f"{self.base}/in.php", data=data, timeout=30,
        )
        result = resp.json()

        if result.get("status") != 1:
            raise RuntimeError(result.get("request"))

        captcha_id = result["request"]
        start = time.time()

        while time.time() - start < timeout:
            time.sleep(5)
            resp = requests.get(f"{self.base}/res.php", params={
                "key": self.api_key,
                "action": "get",
                "id": captcha_id,
                "json": 1,
            }, timeout=15)
            data = resp.json()
            if data["request"] != "CAPCHA_NOT_READY":
                if data.get("status") == 1:
                    return data["request"]
                raise RuntimeError(data["request"])

        raise TimeoutError("Solve timeout")

    def stats(self):
        return {
            "workers": len(self.workers),
            "active": self.active_count,
            "queue": self.redis.llen(self.queue_key),
        }


# Usage
pool = AutoScalingPool(os.environ["CAPTCHAAI_KEY"])
pool.start()

# Monitor
while True:
    print(pool.stats())
    time.sleep(30)

لاحظ قيمة max_workers = 20: هذا هو السقف الفعلي لتزامنك، ويجب ألّا يتجاوز عدد الخيوط المتاح في خطتك. أي خيط زائد عن حدّ الخطة سينتظر دوره بلا فائدة.

التوسع التلقائي على مستوى العمليات

عندما يرافق الحلّ معالجةٌ ثقيلة مقيّدة بوحدة المعالجة المركزية — كتحضير الصور قبل إرسالها — تصبح العمليات المنفصلة أفضل من الخيوط لأنها تعزل استهلاك المعالج وتتجاوز قيد المُفسّر العالمي في Python. المُقاس التالي يضبط عدد العمليات وفق عمق قائمة الانتظار مع تنظيف العمليات المنتهية:

import multiprocessing
import time
import redis
import os


class ProcessScaler:
    """Scale worker processes based on queue depth."""

    def __init__(self, worker_fn, redis_url="redis://localhost:6379"):
        self.worker_fn = worker_fn
        self.redis = redis.from_url(redis_url)
        self.processes = []
        self.min_workers = 2
        self.max_workers = 16

    def run(self, check_interval=15):
        """Run the scaler loop."""
        # Start minimum workers
        for _ in range(self.min_workers):
            self._spawn()

        while True:
            time.sleep(check_interval)
            self._cleanup_dead()

            queue_depth = self.redis.llen("captcha:tasks")
            current = len(self.processes)

            # Scale up
            if queue_depth > current * 5 and current < self.max_workers:
                to_add = min(
                    max(1, queue_depth // 10),
                    self.max_workers - current,
                )
                for _ in range(to_add):
                    self._spawn()
                print(f"Scaled up to {len(self.processes)} workers")

            # Scale down
            elif queue_depth < 3 and current > self.min_workers:
                to_remove = min(2, current - self.min_workers)
                for _ in range(to_remove):
                    p = self.processes.pop()
                    p.terminate()
                print(f"Scaled down to {len(self.processes)} workers")

    def _spawn(self):
        p = multiprocessing.Process(target=self.worker_fn)
        p.start()
        self.processes.append(p)

    def _cleanup_dead(self):
        self.processes = [p for p in self.processes if p.is_alive()]
        # Ensure minimum
        while len(self.processes) < self.min_workers:
            self._spawn()

استدعاء _cleanup_dead() في كل دورة يمنع تراكم عمليات «الزومبي» التي انتهت دون تنظيف ويحافظ على الحدّ الأدنى قائمًا دائمًا.

إيقاف التوسع عند انخفاض الرصيد

قبل رفع أي طاقة، تحقّق من الرصيد؛ فإضافة عمّال والرصيد يقترب من الصفر تعني مهامًا فاشلة وسجلًّا مليئًا بالأخطاء. الدالة التالية تستطلع getbalance وتُرجع قرارًا واضحًا:

def check_balance(api_key, min_balance=2.0):
    """Check if balance is sufficient for scaling."""
    resp = requests.get("https://ocr.captchaai.com/res.php", params={
        "key": api_key,
        "action": "getbalance",
        "json": 1,
    }, timeout=15)
    balance = float(resp.json()["request"])

    if balance < min_balance:
        print(f"Balance ${balance:.2f} below ${min_balance} — halting scale-up")
        return False
    return True

ثم اربط الفحص بشرط الرفع داخل حلقة القياس ليتوقّف التوسع دون الحدّ الآمن:

# In _scaling_loop:
if queue_depth > 20 and utilization > 70:
    if check_balance(self.api_key, min_balance=2.0):
        # Scale up
        ...
    else:
        print("Scaling paused — low balance")

مواءمة حدود التوسع مع خطة CaptchaAI

هنا يظهر أثر نموذج التسعير المبني على الخيوط بوضوح: السقف المنطقي لـ max_workers هو عدد خيوط خطتك، لأن التزامن الفعلي محدود بالخيوط لا بعدد العمّال في الكود. جميع الخطط تشمل عددًا غير محدود من الحلول لكل خيط شهريًا، فلا رسوم لكل اختبار CAPTCHA ولا سقوف يومية — يبقى المتغيّر الوحيد هو كم خيطًا تشغّل بالتوازي.

مثال max_workers = 20 في الكود أعلاه يناسب خطة ADVANCE ‏(‎$90 شهريًا، 50 خيطًا) بهامش وافر، ويمكن رفعه حتى 50 على الخطة نفسها. للاطلاع على الطاقات الأعلى راجع صفحة الأسعار الرسمية على captchaai.com؛ رفع max_workers فوق خيوط خطتك لا يزيد الإنتاجية بل يترك خيوطًا معلّقة في الانتظار.

وفي ضبط هذا السقف نقطتان:

• التزامن الفعلي هو أقلّ قيمة بين max_workers وخيوط خطتك.
• رفع max_workers وحده لا يزيد الطاقة؛ الترقية إلى خطة أعلى هي ما يفعل.

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

اختيار الأسلوب يوازن بين البساطة والعزل وطبيعة النشر لديك:

ابدأ بمجمّع الخيوط لبساطته، وانتقل إلى العمليات أو Kubernetes فقط عندما تفرض المعالجة الثقيلة أو التوزيع على عدة عُقد ذلك.

معالجة المشكلات الشائعة

يربط الجدول التالي أكثر أعطال التوسع شيوعًا بسببها المباشر وإجراء التصحيح:

قائمة تحقّق قبل التشغيل في الإنتاج

راجع هذه النقاط قبل ربط المُقاس بحمل حقيقي:

• max_workers مضبوط عند عدد خيوط خطتك أو أقل.
• فحص الرصيد مفعّل قبل كل رفع للطاقة.
• تأخير التقليص أطول من تأخير الرفع لكبح التذبذب.
• تنظيف العمليات المنتهية مجدول في كل دورة قياس.

الأسئلة الشائعة

كيف أواءم عدد العمّال مع خيوط خطتي في CaptchaAI؟

اجعل max_workers مساويًا لعدد خيوط خطتك أو أقل. مثلًا خطة ADVANCE ‏(‎$90، 50 خيطًا) تتحمّل حتى 50 عاملًا متزامنًا؛ أي عدد أعلى يترك خيوطًا في الانتظار دون زيادة في الإنتاجية.

هل يستهلك التوسع الرصيد أسرع؟

التزامن الأعلى يعني حلولًا أكثر في الوحدة الزمنية، والفوترة قائمة على الخيوط لا على كل حل. لذا احرص على فحص الرصيد قبل كل رفعٍ للطاقة عبر getbalance، وأوقف التوسع تلقائيًا دون الحدّ الآمن الذي تحدّده.

ما الإشارة الأهم لاتخاذ قرار التوسع؟

عمق قائمة الانتظار مقترنًا بنسبة انشغال العمّال. طابور طويل مع عمّال مشغولين يبرّر الرفع؛ أما طابور طويل مع عمّال خاملين فيشير إلى عطل يجب فحصه قبل إضافة أي طاقة.

كيف أتجنّب التذبذب المتكرر في العدد؟

افصل بين سرعتي الرفع والخفض: افحص كل 10 إلى 15 ثانية للرفع، وانتظر 30 إلى 60 ثانية من الحمل المنخفض قبل الخفض. هذا الفارق يمنع تأرجح النظام بين حالتي التوسع والتقليص.

أدلة ذات صلة

• مراقبة مواقع المزادات والتعامل مع CAPTCHA
• التعامل مع اللغة العربية وتحدّيات RTL في CAPTCHA

جاهز للتشغيل بذكاء؟ احصل على مفتاح CaptchaAI وابدأ بربط حدود التوسع بخيوط خطتك اليوم.