CaptchaAI Ratenbegrenzungen und Drosselung

CaptchaAI unterstützt hohe Anforderungsvolumina, aber wenn Sie die Kapazitätsgrenzen kennen und eine ordnungsgemäße Drosselung implementieren, bleibt Ihre Pipeline zuverlässig.

API-Ratenbegrenzungen

CaptchaAI erzwingt keine strengen Ratenbegrenzungen pro Sekunde. Stattdessen wird die Kapazität durch die Verfügbarkeit der Arbeitskräfte verwaltet:

Die Hauptbeschränkung ist die Arbeitskapazität. Wenn alle Arbeiter beschäftigt sind, erhalten Sie ERROR_NO_SLOT_AVAILABLE.

Behandlung von ERROR_NO_SLOT_AVAILABLE

Dieser Fehler bedeutet, dass das System ausgelastet ist. Exponentielles Backoff implementieren:

Python

import time
import requests

API_KEY = "YOUR_API_KEY"


def submit_with_backoff(params, max_retries=5):
    params["key"] = API_KEY

    for attempt in range(max_retries):
        resp = requests.get(
            "https://ocr.captchaai.com/in.php", params=params
        )

        if resp.text.startswith("OK|"):
            return resp.text.split("|")[1]

        if resp.text == "ERROR_NO_SLOT_AVAILABLE":
            wait = min(2 ** attempt * 2, 60)  # 2, 4, 8, 16, 32s max 60
            print(f"No slots, waiting {wait}s (attempt {attempt + 1})")
            time.sleep(wait)
            continue

        raise Exception(f"Submit error: {resp.text}")

    raise Exception("Max retries exceeded — no slots available")

Node.js

async function submitWithBackoff(params, maxRetries = 5) {
  params.key = process.env.CAPTCHAAI_API_KEY;

  for (let attempt = 0; attempt < maxRetries; attempt++) {
    const resp = await axios.get("https://ocr.captchaai.com/in.php", {
      params,
    });
    const text = String(resp.data);

    if (text.startsWith("OK|")) {
      return text.split("|")[1];
    }

    if (text === "ERROR_NO_SLOT_AVAILABLE") {
      const wait = Math.min(2 ** attempt * 2000, 60000);
      console.log(`No slots, waiting ${wait}ms (attempt ${attempt + 1})`);
      await new Promise((r) => setTimeout(r, wait));
      continue;
    }

    throw new Error(`Submit error: ${text}`);
  }

  throw new Error("Max retries exceeded");
}

Clientseitige Ratenbegrenzung

Token-Bucket (Python)

import time
import threading


class RateLimiter:
    def __init__(self, rate, per=1.0):
        """Allow `rate` requests per `per` seconds."""
        self.rate = rate
        self.per = per
        self.tokens = rate
        self.last_refill = time.monotonic()
        self.lock = threading.Lock()

    def acquire(self):
        with self.lock:
            now = time.monotonic()
            elapsed = now - self.last_refill
            self.tokens = min(self.rate, self.tokens + elapsed * (self.rate / self.per))
            self.last_refill = now

            if self.tokens >= 1:
                self.tokens -= 1
                return
            else:
                sleep_time = (1 - self.tokens) * (self.per / self.rate)

        time.sleep(sleep_time)
        self.acquire()


# Allow 10 submissions per second
limiter = RateLimiter(rate=10, per=1.0)

def submit_limited(params):
    limiter.acquire()
    return submit_with_backoff(params)

Asyncio-Semaphor

import asyncio

# Limit to 20 concurrent tasks
semaphore = asyncio.Semaphore(20)

async def solve_limited(solver, session, params):
    async with semaphore:
        return await solver.solve(session, params)

Poll-Rate-Kontrolle

Führen Sie eine Umfrage nicht häufiger als alle 5 Sekunden pro Aufgabe durch:

async def smart_poll(session, task_id, solver):
    """Polls with adaptive intervals."""
    intervals = [5, 5, 5, 10, 10, 15, 15, 30, 30, 60]

    for wait in intervals:
        await asyncio.sleep(wait)
        result = await solver.check(session, task_id)
        if result is not None:
            return result

    raise TimeoutError(f"Task {task_id} timed out")

Überwachung und Metriken

Verfolgen Sie Ihre API-Nutzung, um innerhalb der Grenzen zu bleiben:

import time
from collections import deque


class APIMetrics:
    def __init__(self, window=60):
        self.window = window
        self.requests = deque()
        self.errors = deque()

    def record_request(self):
        now = time.time()
        self.requests.append(now)
        self._cleanup(self.requests, now)

    def record_error(self, error_code):
        now = time.time()
        self.errors.append((now, error_code))
        self._cleanup_tuples(self.errors, now)

    def get_rate(self):
        now = time.time()
        self._cleanup(self.requests, now)
        return len(self.requests) / self.window

    def get_error_rate(self):
        now = time.time()
        self._cleanup(self.requests, now)
        self._cleanup_tuples(self.errors, now)
        if not self.requests:
            return 0
        return len(self.errors) / len(self.requests)

    def _cleanup(self, dq, now):
        while dq and dq[0] < now - self.window:
            dq.popleft()

    def _cleanup_tuples(self, dq, now):
        while dq and dq[0][0] < now - self.window:
            dq.popleft()


metrics = APIMetrics()

# Use in your submit function
def submit_tracked(params):
    metrics.record_request()
    try:
        return submit_with_backoff(params)
    except Exception as e:
        metrics.record_error(str(e))
        raise

# Check metrics periodically
print(f"Rate: {metrics.get_rate():.1f} req/s")
print(f"Error rate: {metrics.get_error_rate():.1%}")

Kapazitätsplanung

Für Volumes über 10K/hour kontaktieren Sie den CaptchaAI-Support für dedizierte Kapazität.

FAQ

Gibt es ein Tageslimit?

Kein festes Tageslimit. Ihr Limit wird durch Ihr Guthaben und die Worker-Kapazität der API bestimmt.

Was passiert, wenn ich die Kapazität überschreite?

Sie erhalten ERROR_NO_SLOT_AVAILABLE. Es ist vorübergehend – versuchen Sie es nach einer Backoff-Zeit erneut.

Sollte ich auf der Clientseite drosseln?

Ja, für vorhersehbares Verhalten. Verwenden Sie eine Semaphore, um gleichzeitige Aufgaben zu begrenzen, und einen Token-Bucket, um die Übermittlungsrate zu steuern.

Verwandte Leitfäden

• Parallele CAPTCHA-Lösung
• CaptchaAI-API-Latenz optimieren
• API-Antwortformate und Fehlercodes