المولدات والاشتمالات غير المتزامنة

طريقان إلى نفس الفكرة

الأساسيات المطلوبة

المفاهيم الأساسية

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

1تذكير سريع: المولّد العادي

الدالة التي تحتوي yield لا تُرجع قيمة، بل تُرجع مولّدًا: كائنًا يُنتج قيمة عند كل طلب ثم يتجمّد عند ذلك السطر حتى الطلب التالي.

def counter():
    for i in range(3):
        yield i

g = counter()           # لم يُنفَّذ شيء بعد
print(next(g), next(g)) # 0 1   كل استدعاء يُشغّل حتى yield التالي

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

2المولّد غير المتزامن: yield وawait في جسم واحد

قبل بايثون 3.6 كان الجمع بينهما ممنوعًا. رفع PEP 525 هذا المنع، فصار بإمكانك كتابة:

async def menu_stream():
    for i in range(5):
        await asyncio.sleep(0.2)      # انتظار يعيد الدور للحلقة
        yield round(random.uniform(0, 10), 2)   # ثم أنتج قيمة

الدالة الآن ليست كوروتينًا رغم async def‏: وجود yield غيّر نوعها بالكامل. استدعاؤها يعطيك كائنًا من نوع async_generator‏، ولا يصحّ أن تكتب await أمامه، بل async for‏.

3الأشكال الأربعة المتقاربة (الجدول الذي يحسم الالتباس)

4async for‏: الحلقة التي تنتظر بين الدورات

الحلقة العادية for تطلب العنصر التالي بـ__next__ وتستلمه فورًا. أما async for فتطلبه بـ__anext__ الذي يُرجع شيئًا يُنتظر، فتعيد الحلقة الدور لحلقة الأحداث بين كل عنصرين. ونهاية التكرار تصل كاستثناء StopAsyncIteration تلتقطه الحلقة نيابة عنك.

وasync for‏، مثل await‏، ممنوعة خارج دالة async‏.

5الاشتمال غير المتزامن

أضاف PEP 530 في بايثون 3.6 السماح بـasync for داخل الاشتمالات، فصار هذا:

result = []
async for v in menu_stream():
    result.append(v)

يُكتب هكذا:

result = [v async for v in menu_stream()]

وتعمل الفكرة في اشتمالات القوائم والمجموعات والقواميس والمولّدات، ومعها شرط اختياري:

big = [v async for v in menu_stream() if v > 5]
as_set = {v async for v in menu_stream()}
as_dict = {i: v async for i, v in numbered_stream()}

6await داخل الاشتمال: حالة مختلفة

انتبه للفرق بين وجهين يسهل الخلط بينهما:

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

7التعليقات النوعية للمولّدات

from typing import AsyncGenerator

async def count_slowly() -> AsyncGenerator[float, None]:
    ...

8إغلاق المولّد والتنظيف

المولّد غير المتزامن قد يكون ممسكًا باتصال أو ملف. إن لم تستهلكه حتى النهاية فأغلقه بنفسك بـawait g.aclose()‏، وضع تنظيفك في finally داخل المولّد ليعمل في الحالتين. وحلقة async for التي تنتهي طبيعيًا تتولى هذا عنك.

9لماذا أربعة اشتمالات متوازية ≈ زمن واحد؟

هذا أهم سؤال في هذا الجزء، وجوابه في سطرين:

1. داخل الاشتمال الواحد: عشر قيم، كل واحدة بعد انتظار ثانية، والانتظارات متتابعة لأن المولّد لا يُنتج القيمة الثانية قبل الأولى. المجموع ≈ 10 ثوانٍ، ولا سبيل لاختصاره.
2. بين الاشتمالات الأربعة: لا علاقة بينها، فحين تمرّرها لـgather تتداخل انتظاراتها الأربعة تمامًا. المجموع ≈ 10 ثوانٍ أيضًا، لا 40.

أي أن asyncio يختصر الانتظارات المستقلة ولا يختصر الانتظارات المتسلسلة. وسنقيس هذا بالأرقام في الخطوة السادسة.

كيف يعمل هذا فعليًا، خطوة بخطوة

ثلاث صور: كيف تميّز الشكل الذي كتبته، وماذا يجري داخل async for‏، ولماذا لا يتضاعف الزمن مع عدد الاشتمالات.

الرسم 40: كيف تعرف ما الذي كتبته. سؤالان فقط يحدّدان النوع وطريقة الاستهلاك، ومعظم أخطاء «‏object is not iterable‏» سببها الخلط بين فرعين من هذه الشجرة.

الرسم 41: ما يجري بين كل عنصرين. الفرق عن for العادية أن فترة الانتظار هذه ليست ضائعة: الحلقة تشغّل غيرك خلالها.

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

1. اكتب المصدر مولّدًا غير متزامن إذا كانت القيم تصل تباعًا مع انتظار بينها.
2. استهلكه بـasync for إن احتجت منطقًا داخل الحلقة، أو باشتمال إن كان الهدف جمع قائمة.
3. علّق النوع بـAsyncGenerator[Y, None] المستورد من typing على 3.8.
4. لتشغيل عدة اشتمالات مستقلة معًا، مرّرها لـgather‏؛ الاشتمال وحده لا يوازي شيئًا.
5. قِس بـperf_counter وتأكّد أن الرقم يطابق توقّعك، وإلا فافهم لماذا قبل أن تمضي.

مثال محلول: شاشة الطلبات التي تتدفّق

في المقهى شاشة تعرض الطلبات كلما وصلت من الكاشير، لا دفعة واحدة في آخر اليوم. كل كود ومخرَج تحته نُفِّذ فعليًا.

الخطوة 1: المصدر، مولّد غير متزامن

import asyncio, random, time
random.seed(11)

async def menu_stream():
    """يُنتج 5 طلبات، بينها 0.2 ثانية انتظار"""
    for i in range(5):
        await asyncio.sleep(0.2)
        yield round(random.uniform(0, 10), 2)

الخطوة 2: استهلاكه بـasync for

async def main():
    t0 = time.perf_counter()
    seen = []
    async for v in menu_stream():
        seen.append(v)
    print("async for  ->", seen, "in", round(time.perf_counter()-t0, 2), "s")

asyncio.run(main())
# async for  -> [4.52, 5.6, 9.24, 4.66, 5.08] in 1.0 s

خمس قيم × 0.2 ثانية = ثانية واحدة. الانتظارات داخل المولّد متسلسلة بطبيعتها.

الخطوة 3: ما نوع هذا الكائن أصلًا؟

import inspect

async def inspect_it():                              # داخل دالة async بسبب aclose
    g = menu_stream()
    print(type(g).__name__)                          # async_generator
    print(inspect.isasyncgenfunction(menu_stream))   # True
    print(inspect.iscoroutinefunction(menu_stream))  # False
    await g.aclose()                                 # إغلاق يدوي لأننا لم نستهلكه

asyncio.run(inspect_it())

النتيجة الثالثة هي المفاجأة المفيدة: المولّد غير المتزامن ليس كوروتينًا، ولهذا لا ينفع معه await ولا asyncio.run مباشرة.

الخطوة 4: نفس العمل باشتمال من سطر واحد

vals = [v async for v in menu_stream()]
print("async comp ->", vals)
# async comp -> [5.87, 1.85, 5.12, 6.3, 7.93]   in 1.0 s

نفس الزمن ونفس النتيجة، بثلاثة أسطر أقل. الاشتمال اختصار للحلقة، لا تسريع لها.

الخطوة 5: مع شرط

big = [v async for v in menu_stream() if v > 5]
print("filtered   ->", big)
# filtered   -> [8.1, 6.93]

الشرط يُطبَّق بعد وصول كل قيمة، تمامًا كالاشتمال العادي.

الخطوة 6: أربعة اشتمالات، بالتتابع ثم بالتوازي

async def gen():                       # 10 قيم، بينها 0.1 ث  -> ثانية لكل تشغيل
    for _ in range(10):
        await asyncio.sleep(0.1)
        yield random.uniform(0, 10)

async def comprehension():
    return [v async for v in gen()]

async def four_sequential():
    t0 = time.perf_counter()
    for _ in range(4):
        await comprehension()          # ننتظر كل واحد قبل التالي
    return round(time.perf_counter() - t0, 2)

async def four_parallel():
    t0 = time.perf_counter()
    await asyncio.gather(*(comprehension() for _ in range(4)))
    return round(time.perf_counter() - t0, 2)

print("4 x sequential:", asyncio.run(four_sequential()), "s")   # 4.02 s
print("4 x gather    :", asyncio.run(four_parallel()), "s")     # 1.0 s

الخطوة 7: فخ «اشتمال يبدو متزامنًا»

jobs = [comprehension for _ in range(4)]      # داخل دالة async

slow = [await f() for f in jobs]                  # 4.02 ث، نفس رقم «بالتتابع» أعلاه
fast = await asyncio.gather(*(f() for f in jobs)) # 1.0  ث، نفس رقم gather أعلاه

السطران يبدوان متقاربين والفرق أربعة أضعاف، وهو نفس الفرق الذي قسناه في الخطوة السادسة لأن الاشتمال هنا ليس إلا كتابة أخرى للحلقة المتتابعة. كلمة await داخل اشتمال تعني «انتظر هذه ثم انتقل»، وهذا تتابع لا تزامن.

الخطوة 8: قياس الزمن داخل كوروتين

async def time_four_runs():
    t0 = time.perf_counter()
    await asyncio.gather(*(comprehension() for _ in range(4)))
    return time.perf_counter() - t0

لاحظ الفرق عن درس البرمجة غير المتزامنة: هنا الدالة كوروتين لأنها تنتظر gather داخلها، بينما كانت هناك دالة عادية لأنها تستدعي asyncio.run‏. القاعدة: من ينتظر شيئًا يكون كوروتينًا، ومن يفتح الحلقة يكون دالة عادية.

أسطورة وحقيقة

أين يُستخدم هذا عمليًا؟

جرّب بنفسك

بطاقات مراجعة سريعة

تحقّق من فهمك