asyncio في Python: البرمجة غير المتزامنة، وTaskGroup، والخيوط، وGIL

asyncio مكتبة مدمجة في Python للكود غير المتزامن. الدوال المعرّفة بـ async def تستطيع التوقف عند await، وأثناء انتظار رد الشبكة ينفذ البرنامج مهاماً أخرى. وهكذا تجري مئات الطلبات معاً في خيط واحد. لكن asyncio أداة واحدة من ثلاث: توجد أيضاً الخيوط (threading) والعمليات (multiprocessing). فيما يلي شرح لطريقة عمل asyncio في Python، وأي أداة تختار لمهمتك، ودور GIL في الموضوع.

الانتظار والحساب مهمتان مختلفتان

من دون هذا التمييز يصعب فهم ما يلي. البرنامج البطيء يتباطأ لأحد سببين:

• مرتبط بالإدخال والإخراج (I/O-bound) — ينتظر: رد API أو قاعدة بيانات أو قراءة قرص. والمعالج في هذه الحالة خامل.
• مرتبط بالمعالج (CPU-bound) — يحسب: يعالج صوراً أو يدور على ملايين السجلات. والمعالج مشغول.

للانتظار تناسب asyncio أو الخيوط: بينما تنتظر عملية تنفذ أخرى. أما للحسابات فتحتاج إلى حساب متوازٍ حقيقي على عدة أنوية، أي العمليات. والخلط بينهما هو الخطأ الرئيسي: الخيوط لا تسرّع الحساب في Python العادي، والعمليات من أجل ثلاثة طلبات HTTP مبالغة لا داعي لها.

ما هو asyncio: الدوال المشتركة وحلقة الأحداث

يقوم الكود غير المتزامن على كلمتين. async def تعرّف دالة مشتركة (coroutine) تستطيع التوقف. واستدعاؤها لا ينفذ جسمها، بل ينشئ كائن الدالة المشتركة. أما await داخلها فتعني: «سأنتظر هنا، فلتتفرغ للمهام الأخرى». ويدير كل ذلك حلقة الأحداث (event loop) التي تُشغَّل عبر asyncio.run().

import asyncio

async def fetch_rate(currency):
    await asyncio.sleep(1)          # محاكاة طلب إلى API
    return {"USD": 92.5, "EUR": 100.1, "CNY": 12.7}[currency]

async def main():
    rate = await fetch_rate("USD")
    print(rate)

asyncio.run(main())  # 92.5

asyncio.sleep() توقف غير متزامن: تعيد التحكم إلى حلقة الأحداث. أما time.sleep() العادية داخل دالة مشتركة فستوقف الحلقة كلها، وتقف بقية المهام.

بالتسلسل أم بالتوازي

إذا كتبت await داخل حلقة، فستنفذ الطلبات واحداً تلو الآخر، فتستغرق العملات الثلاث ثلاث ثوانٍ. ولكي تجري معاً يجب تشغيلها كمهام.

import asyncio, time

async def main():
    start = time.perf_counter()
    for cur in ("USD", "EUR", "CNY"):
        await fetch_rate(cur)                       # واحدة تلو الأخرى
    print(f"{time.perf_counter() - start:.1f} ثانية")   # 3.0 ثانية

asyncio.run(main())

TaskGroup: تشغيل معاً والانتظار للجميع

الطريقة الحديثة (منذ Python 3.11) هي مجموعة المهام. داخل كتلة async with تبدأ كل مهمة فوراً عبر tg.create_task()، والخروج من الكتلة يحدث عندما تنتهي كل المهام.

async def main():
    start = time.perf_counter()
    async with asyncio.TaskGroup() as tg:
        tasks = {cur: tg.create_task(fetch_rate(cur)) for cur in ("USD", "EUR", "CNY")}
    rates = {cur: t.result() for cur, t in tasks.items()}
    print(rates, f"{time.perf_counter() - start:.1f} ثانية")   # ... 1.0 ثانية

asyncio.run(main())

تعمل TaskGroup بصرامة: إذا سقطت مهمة واحدة تُلغى البقية، وتُجمع كل الأخطاء في ExceptionGroup. ويُلتقط هذا بـ except*:

async def main():
    try:
        async with asyncio.TaskGroup() as tg:
            tg.create_task(fetch_rate("USD"))
            tg.create_task(fetch_rate("XXX"))   # KeyError
    except* KeyError as eg:
        print("عملة غير معروفة:", eg.exceptions)

asyncio.gather وcreate_task

قبل TaskGroup كان المعيار هو asyncio.gather(): يشغّل الدوال المشتركة معاً ويعيد النتائج بالترتيب نفسه. ولا يزال كثير منه موجوداً في الكود القديم.

async def main():
    usd, eur, cny = await asyncio.gather(
        fetch_rate("USD"), fetch_rate("EUR"), fetch_rate("CNY")
    )
    # مع return_exceptions=True تعود الأخطاء في القائمة بدلاً من إيقاف الكل

تشغّل asyncio.create_task() دالة مشتركة في الخلفية دون مجموعة. ويجب حفظ مرجع للمهمة، وإلا فقد يحذفها جامع المهملات، ولن يرى أحد أخطاءها. لذلك تبقى TaskGroup أكثر أماناً للمشاريع الجديدة.

إلغاء المهام والمهلات الزمنية

async def main():
    task = asyncio.create_task(fetch_rate("USD"))
    task.cancel()                     # سيُرفع CancelledError داخل المهمة
    try:
        await task
    except asyncio.CancelledError:
        print("أُلغيت")

    try:
        async with asyncio.timeout(0.5):   # منذ Python 3.11
            await fetch_rate("EUR")
    except TimeoutError:
        print("استغرق وقتاً طويلاً")

في الإصدارات القديمة تُستخدم asyncio.wait_for(coro, timeout=0.5) بدل asyncio.timeout(). وتوجد أيضاً asyncio.wait()، وهي تعيد مجموعتين: المهام المنتهية وغير المنتهية، ويمكنها انتظار الأولى (return_when=asyncio.FIRST_COMPLETED).

وفي Python 3.15، كما يظهر في الدورة، أُضيفت إلى TaskGroup الطريقة tg.cancel(): تلغي كل مهام المجموعة دون استثناء. وهذا يغطي حالة «من يجيب أولاً»: تسأل ثلاث خدمات بديلة، تأخذ أول رد، وتلغي الباقي. وكان يُفعل ذلك سابقاً برفع استثناء إشارة خاص أو بإلغاء المهام يدوياً.

التزامن: Semaphore وLock وEvent وQueue

لـ asyncio أدوات تزامن خاصة تعمل مع await. وأنفعها Semaphore: تحدد عدد العمليات المتزامنة، حتى لا تُغرق API غيرك بآلاف الطلبات.

sem = asyncio.Semaphore(10)

async def limited(url):
    async with sem:              # لا أكثر من 10 في وقت واحد
        await asyncio.sleep(0.1) # هنا كان سيكون الطلب
        return url

• asyncio.Lock — دالة مشتركة واحدة فقط داخل القسم الحرج.
• asyncio.Event — دالة تنتظر await event.wait()، وأخرى تُطلق الإشارة event.set().
• asyncio.Queue — طابور «منتِج ← مستهلك»: await q.put() وawait q.get() وq.task_done().

فخ: نسيان await

أكثر الأخطاء شيوعاً هو استدعاء دالة مشتركة دون await:

async def main():
    rate = fetch_rate("USD")    # نسينا await
    print(rate * 2)             # TypeError
# RuntimeWarning: coroutine 'fetch_rate' was never awaited

لم ينفذ Python الدالة، بل أنشأ كائن الدالة المشتركة فقط. فإذا رأيت في السجلات الرسالة «was never awaited» فابحث عن await ناقص. والخطأ الثاني الشائع هو الاستدعاء المانع (time.sleep أو requests.get) داخل دالة مشتركة. ولطلبات HTTP استخدم مكتبات غير متزامنة (مثل httpx وaiohttp)، والكود المتزامن الذي لا مفر منه انقله إلى خيط عبر await asyncio.to_thread(func, arg).

تعدد الخيوط في Python: الوحدة threading

الخيط (thread) مسار تنفيذ مستقل داخل العملية الواحدة، والخيوط تتشارك الذاكرة. وهي تعمل جيداً في انتظار الشبكة، ولا تتطلب إعادة كتابة الكود بصيغة async.

import threading, time

def download(name):
    time.sleep(1)               # انتظار الشبكة
    print(name, "تم")

threads = [threading.Thread(target=download, args=(n,)) for n in ("a", "b", "c")]
for t in threads:
    t.start()
for t in threads:
    t.join()                    # الكل في نحو ثانية واحدة

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

ThreadPoolExecutor

تنشئ concurrent.futures.ThreadPoolExecutor مجمّع خيوط وتوزع عليه المهام. وتعيد map النتائج بالترتيب الأصلي.

from concurrent.futures import ThreadPoolExecutor
import time

def fetch(url):
    time.sleep(1)
    return f"{url}: 200"

urls = [f"https://example.com/{i}" for i in range(5)]
with ThreadPoolExecutor(max_workers=5) as pool:
    for result in pool.map(fetch, urls):
        print(result)           # كله في نحو ثانية واحدة

ما هو GIL ولماذا لا تسرّع الخيوط الحسابات

GIL (Global Interpreter Lock) قفل عام في مفسّر CPython. ففي كل لحظة لا ينفذ بايت كود Python إلا خيط واحد. وعندما ينتظر الخيط الإدخال والإخراج يفرج عن GIL، لذلك تفيد الخيوط في الشبكة. أما الحساب الصرف فتعمل فيه أربعة خيوط لا أسرع من خيط واحد، لأنها تتناوب على القفل.

from concurrent.futures import ThreadPoolExecutor

def heavy(n):
    return sum(i * i for i in range(n))

with ThreadPoolExecutor(4) as pool:
    list(pool.map(heavy, [5_000_000] * 4))   # تقريباً مثل التنفيذ التسلسلي

المكتبات المكتوبة بلغة C (مثل NumPy وما شابهها) تفرج عن GIL أثناء العمليات الثقيلة في الغالب، لذلك تحقق الخيوط معها تسريعاً.

multiprocessing: التوازي الحقيقي

لكل عملية مفسّرها وGIL الخاص بها، فتحسب العمليات على أنوية مختلفة في الوقت نفسه. والثمن ذاكرة منفصلة: تُنقل البيانات بين العمليات بالتسلسل (pickle)، وتكلفة تشغيل عملية أعلى من تشغيل خيط.

from concurrent.futures import ProcessPoolExecutor

def heavy(n):
    return sum(i * i for i in range(n))

if __name__ == "__main__":      # ضروري على Windows وmacOS
    with ProcessPoolExecutor() as pool:
        print(list(pool.map(heavy, [5_000_000] * 4)))

من دون التحقق if __name__ == "__main__" ستبدأ العمليات الفرعية عند استيراد الوحدة بإنشاء عمليات جديدة. وتوجد وحدة منخفضة المستوى multiprocessing فيها Process وPool وQueue، لكن ProcessPoolExecutor أبسط ولها الواجهة نفسها لمجمّع الخيوط.

ماذا تختار: asyncio أم threading أم multiprocessing

Python بلا GIL والمفسّرات الفرعية: الجديد

منذ Python 3.13 توجد نسخة بناء بلا GIL (free-threading، PEP 703) تُعرف بحرف t في الاسم، مثل python3.15t. كانت تجريبية في 3.13، ومنذ 3.14 (PEP 779) صارت مدعومة رسمياً. تحسب الخيوط فيها بالتوازي على أنوية مختلفة. لكن افتراضياً ستحصل على Python العادي المزود بـ GIL: الكود أحادي الخيط يعمل فيها أبطأ قليلاً، وليست كل المكتبات جاهزة لها.

import sys
print(sys._is_gil_enabled())   # False في نسخة free-threaded

والطريق الثاني للحسابات المتوازية ظهر في 3.14 (PEP 734): المفسّرات الفرعية (subinterpreters) في الوحدة concurrent.interpreters، وهي عدة مفسّرات معزولة داخل عملية واحدة، لكل منها GIL خاص بها. وأسهل طريقة لاستخدامها هي InterpreterPoolExecutor، وشكله كشكل مجمّع الخيوط. وفي قياس الدورة أنهت المفسّرات الفرعية أربع مهام ثقيلة في 0.39 ثانية، مقابل 1.03 ثانية في خيط واحد.

الخلاصة

asyncio في Python أداة لمهام الانتظار: دوال مشتركة بـ async def، وتوقف بـ await، وتشغيل متزامن عبر TaskGroup أو gather. وتحل الخيوط المسألة نفسها دون إعادة كتابة الكود بصيغة async، أما الحسابات الثقيلة فتحتاج إلى العمليات، لأن GIL يمنع خيوط CPython العادي من الحساب المتوازي. وتغيّر نسخة free-threading والمفسّرات الفرعية هذه الصورة تدريجياً، لكن القاعدة الأساسية ما تزال قائمة: الانتظار يُحل بـ asyncio أو الخيوط، والحساب بالعمليات. وقد شُرح هذا كله عملياً في الدورة: يحصل متتبّع المصروفات على أسعار العملات بالتوازي في ثانية واحدة بدلاً من ثلاث.