تدوير البروكسي غير المتزامن في بايثون: httpx وaiohttp (دليل 2026)

نُشر في 12 يونيو 2026 · ≈10 دقيقة قراءة

الاستخراج غير المتزامن (Async) هو المكان الذي تنهار فيه إعدادات البروكسي بهدوء. مع requests تُدوِّر عنوان IP واحدًا لكل طلب وتمضي قدمًا؛ ومع asyncio تجد فجأة 200 طلب قيد التنفيذ، كل منها يحتاج إلى مخرجه الخاص، وإلى ارتباط الجلسة الخاص به، وإلى إعادة المحاولة الخاصة به عندما يُرجع الهدف رمز 403. أخطئ في ذلك فإما أن تطرق عنوان IP واحدًا حتى يُحظر أو أن تُمزّق كل جلسة تسجيل دخول بالتدوير في منتصف العملية.

يعرض هذا الدليل الأنماط التي تصمد فعليًا عند التزامن: التدوير لكل طلب في httpx وaiohttp، ومجمع عمال محدود، وجلسات sticky للعمليات ذات الحالة، وإعادة المحاولة الواعية بالحظر — مع كود عملي يمكنك لصقه.

لماذا يُغيّر Async مشكلة البروكسي

المستخرِج المتزامن يتعامل مع بروكسي واحد في كل مرة، لذا فإن "التدوير عند كل طلب" صحيح بداهةً. أما Async فيكسر ثلاثة افتراضات دفعة واحدة:

httpx: تدوير البروكسي لكل طلب

يربط httpx.AsyncClient بروكسيًا واحدًا لكل عميل، لذا فللتدوير تختار البروكسي لكل طلب وتُوجّهه عبر مجمع صغير من العملاء مفهرس حسب المخرج:

import asyncio, itertools, httpx

PROXIES = [
    "socks5h://USERNAME:[email protected]:913",
    "socks5h://USERNAME:[email protected]:913",
    "socks5h://USERNAME:[email protected]:913",
]
pool = itertools.cycle(PROXIES)

# One reusable client per exit (connection pooling stays intact)
clients = {p: httpx.AsyncClient(proxy=p, timeout=30) for p in PROXIES}

async def fetch(url):
    proxy = next(pool)
    r = await clients[proxy].get(url)
    return r.status_code, r.text

async def main(urls):
    results = await asyncio.gather(*(fetch(u) for u in urls))
    for client in clients.values():
        await client.aclose()
    return results

إعادة استخدام عميل واحد لكل مخرج يُبقي HTTP/2 وتجميع الاتصالات حيّين بدلًا من دفع ثمن مصافحة جديدة في كل طلب. لاحظ أن httpx>=0.28 أعاد تسمية proxies= إلى proxy= على العميل.

aiohttp: التدوير عبر الطلب

aiohttp أبسط هنا — تأخذ جلسة ClientSession واحدة وسيطًا proxy= لكل طلب، لذا تُدوِّر مضمَّنًا:

import asyncio, itertools, aiohttp

pool = itertools.cycle(PROXIES)  # http:// or socks5h:// (needs aiohttp-socks for SOCKS)

async def fetch(session, url):
    proxy = next(pool)
    async with session.get(url, proxy=proxy, timeout=aiohttp.ClientTimeout(total=30)) as r:
        return r.status, await r.text()

async def main(urls):
    async with aiohttp.ClientSession() as session:
        return await asyncio.gather(*(fetch(session, u) for u in urls))

يتحدث aiohttp بروتوكول بروكسيات HTTP أصليًا؛ ولأجل socks5h:// أضِف aiohttp-socks ومرِّر ProxyConnector.

حُدّ من التزامن بـ Semaphore

gather غير المحدود هو أسرع طريقة لجعل كل عنوان IP في مجمعك يُوسَم دفعة واحدة. حُدّه بحيث يحمل كل مخرج عددًا معقولًا من الطلبات المتوازية يبدو بشريًا:

sem = asyncio.Semaphore(10)   # at most 10 in flight

async def fetch_capped(session, url):
    async with sem:
        return await fetch(session, url)

قاعدة عامة: أبقِ التزامن عند عدد المخارج المتمايزة لديك أو دونه، حتى لا تُكدِّس عدة طلبات متوازية على عنوان IP سكني واحد.

جلسات Sticky للعمليات ذات الحالة

التدوير لأجل الدخول؛ وجلسات sticky لأجل البقاء بالداخل. أي تسجيل دخول، أو عربة تسوق، أو أي شيء مرتبط بملفات تعريف الارتباط يجب أن يثبت على مخرج واحد طوال التسلسل بأكمله. ثبِّت البروكسي لكل مهمة بدلًا من كل طلب:

async def run_account(account, proxy):
    # Same exit IP for every step of this account's flow
    async with httpx.AsyncClient(proxy=proxy, timeout=30) as client:
        await client.post("https://target.site/login", data=account.creds)
        await client.get("https://target.site/dashboard")
        await client.post("https://target.site/cart", json=account.order)

async def main(accounts):
    # One sticky exit per account, accounts run concurrently
    await asyncio.gather(*(run_account(a, p)
                           for a, p in zip(accounts, itertools.cycle(PROXIES))))

استخدم مزوّدًا يدعم جلسات sticky السكنية بحيث يثبت المخرج نفسه طوال عمر المهمة، لا أن يُدوَّر بصمت من تحتك.

إعادة المحاولة الواعية بالحظر

عند التزامن لا يمكنك الوثوق برمز 200 — تُرجع أنظمة مكافحة الروبوتات رمز 200 مع جسم تحدٍّ. اكتشف الحظر وأعِد محاولة المهمة الفاشلة الوحيدة عبر مخرج جديد:

BLOCK_MARKERS = ("just a moment", "/cdn-cgi/challenge-platform",
                 "datadome", "px-captcha", "access denied")

def looks_blocked(status, text):
    return status in (403, 429, 503) or any(m in text[:4000].lower() for m in BLOCK_MARKERS)

async def fetch_retry(url, attempts=4):
    for _ in range(attempts):
        proxy = next(pool)
        async with httpx.AsyncClient(proxy=proxy, timeout=30) as c:
            r = await c.get(url)
            if not looks_blocked(r.status_code, r.text):
                return r
        await asyncio.sleep(0.5)
    raise RuntimeError(f"blocked after {attempts} exits: {url}")

لا يزال يجب أن تتطابق البصمة

مشكلة واحدة لا يحلّها Async: يعتمد كل من httpx وaiohttp على حزمة TLS الخاصة ببايثون، لذا يُرسلان بصمة JA3 لا ينتجها أي متصفح حقيقي. مجمع تدوير مثالي على عنوان IP سكني نظيف لا يزال يُوسَم كأتمتة عند المصافحة. اقرن التدوير غير المتزامن بانتحال TLS — انظر تجاوز بصمة TLS باستخدام curl_cffi، وتحقق من بصمة JA3 وASN لمخرجك على www.jibaoproxy.com/tools/fingerprint.html.

قائمة التحقق

لجميع منتجات الـ IP · مجموعة ضخمة من العُقد المتاحة في أي وقت

سجّل الآن واحصل على ما يصل إلى 100% استرداد على إعادة الشحن

يحصل المستخدمون الجدد على 500MB عند التسجيل، بالإضافة إلى مكافأة على أول عملية شحن. العرض لفترة محدودة.

إنشاء حساب