آموزش ماژول asyncio

درک نیازمندی به asyncio

فرض کنید یک شرکت خدمات وب دارید با دویست سرور که باید مدام وضعیتشان را بررسی کنید. هر پنج ثانیه یک بار باید به هر سرور یک درخواست سلامت (health check) بفرستید و نتیجه را در پایگاه‌داده ثبت کنید. هر درخواست بسته به بار سرور مقصد، حدود نیم تا دو ثانیه طول می‌کشد.

ساده‌ترین روش این است که یک حلقه بنویسیم که به‌ترتیب به تک‌تک سرورها درخواست بزند:

import time

def health_check(server):
    time.sleep(1)          # each request takes ~1s (blocking)
    return "ok"

servers = range(200)
start = time.time()
for s in servers:
    health_check(s)
print("took %.0fs" % (time.time() - start))   # ~200 seconds!

import time

def health_check(server):
    time.sleep(1)          # each request takes ~1s (blocking)
    return "ok"

servers = range(200)
start = time.time()
for s in servers:
    health_check(s)
print("took %.0fs" % (time.time() - start))   # ~200 seconds!

اگر هر درخواست به‌طور میانگین یک ثانیه طول بکشد، کل یک چرخه حدود دویست ثانیه می‌شود. یعنی نمی‌توانیم هر پنج ثانیه وضعیت را به‌روز کنیم، چون خودِ یک دور کامل دویست ثانیه طول می‌کشد. عملاً سامانه بی‌فایده است.

چرا تردها به‌تنهایی کافی نیستند؟

اولین ایده استفاده از تردهاست: به‌ازای هر سرور یک ترد بسازیم و همه را هم‌زمان اجرا کنیم؛ آن‌گاه کل زمان به اندازه‌ی طولانی‌ترین درخواست (حداکثر دو ثانیه) کاهش می‌یابد. مشکل حل شد؟ نه کاملاً. اگر تعداد سرورها از دویست به بیست‌هزار برسد، ساخت بیست‌هزار ترد هم‌زمان عملاً ناممکن است: هر ترد در پایتون حافظه‌ی قابل‌توجهی (چند مگابایت برای پشته) مصرف می‌کند و سیستم‌عامل هم نمی‌تواند این‌همه ترد را مدیریت کند. به‌علاوه، هماهنگی بین تردها و جمع‌آوری نتایج خودش مسئله‌ی جدیدی می‌سازد.

پس با یک چالش واقعی روبه‌رو هستیم: نیاز به ده‌ها هزار عملیات ورودی/خروجی هم‌زمان، بدون امکان استفاده از تردهای فراوان. اینجا جایی است که asyncio وارد می‌شود.

asyncio چیست و چطور کار می‌کند؟

asyncio فقط از یک ترد استفاده می‌کند، اما با آن می‌توان هزاران کار را به‌صورت هم‌روند اجرا کرد. راز کار این است که وقتی یک کار منتظر چیزی است (مثل پاسخ شبکه)، به‌جای بیکار ماندن، کنترل را رها می‌کند تا کار دیگری پیش برود. asyncio از چهار جزء اصلی تشکیل شده است:

• هم‌روال (coroutine): تابعی که می‌تواند در میانه‌ی اجرا متوقف شود و بعداً از همان نقطه، بدون از دست‌دادن وضعیت، ادامه یابد. با async def تعریف می‌شود.
• حلقه‌ی رویداد (event loop): ناظری که مدام بین کارها می‌گردد، کارهای آماده را اجرا می‌کند و کارهای منتظرِ رویداد را کنار می‌گذارد. وقتی هیچ کار آماده‌ای نباشد، به سیستم‌عامل می‌گوید او را بخواباند تا رویداد بعدی برسد.
• تسک (Task): هم‌روالی که در event loop ثبت شده تا اجرا شود. هر Task می‌تواند در حال اجرا، آماده، یا در انتظار باشد.
• منتظرشدن (awaiting): عملی که با آن هم‌روال جاری می‌گوید «تا وقتی چیزی که منتظرش هستم فراهم نشود نمی‌توانم ادامه دهم» و کنترل را به event loop بازمی‌گرداند تا در این فرصت کارهای دیگر اجرا شوند.

مثال – سفارش تلفنی صبحانه

می‌خواهید چند قلم صبحانه را تلفنی سفارش دهید: به نانوایی زنگ می‌زنید و نان سفارش می‌دهید، به لبنیاتی برای شیر، به میوه‌فروشی برای خیار و گوجه، و به سوپرمارکت برای آب‌پرتقال. پس از ثبت همه‌ی سفارش‌ها دیگر کار فعالی ندارید؛ روی مبل دراز می‌کشید و می‌خوابید — اما نه آن‌قدر عمیق که صدای زنگ در را نشنوید. هر بار که پیک یک سفارش می‌رسد، بیدار می‌شوید، تحویل می‌گیرید و دوباره می‌خوابید. وقتی هر چهار سفارش رسید، بیدار می‌مانید و صبحانه را آماده می‌کنید.

نکته‌ی کلیدی: شما فقط بعد از ثبت سفارش‌ها می‌خوابید، نه قبل از آن. به همین ترتیب، حلقه‌ی رویداد فقط زمانی به خواب می‌رود که هیچ Task آماده‌ای در صف نباشد. تا وقتی حداقل یک کار آماده هست، حلقه هرگز نمی‌خوابد و مدام کار می‌کند؛ فقط وقتی واقعاً کاری برای انجام نیست، پردازنده را به سیستم‌عامل واگذار می‌کند تا او را برای رویداد بعدی بیدار کند. برخلاف time.sleep که ترد را بی‌فایده بلاک می‌کند، این «خواب» هیچ CPU مصرف نمی‌کند.

اولین کد asyncio

حالا همان مثال صبحانه را در کد ببینیم. با async def یک هم‌روال می‌سازیم، با await asyncio.sleep(...) «منتظر رسیدن سفارش» می‌مانیم (بدون بلاک‌کردن)، و با asyncio.run(...) حلقه‌ی رویداد را اجرا می‌کنیم:

import asyncio
import time

async def order(item, seconds):
    print(f"ordering {item}...")
    await asyncio.sleep(seconds)     # non-blocking wait (like waiting for the delivery)
    print(f"{item} delivered")
    return item

async def main():
    start = time.time()
    # run all orders concurrently and collect the results
    results = await asyncio.gather(
        order("bread", 2),
        order("milk", 1),
        order("juice", 3),
    )
    print("all delivered:", results, "in %.1fs" % (time.time() - start))

asyncio.run(main())

import asyncio
import time

async def order(item, seconds):
    print(f"ordering {item}...")
    await asyncio.sleep(seconds)     # non-blocking wait (like waiting for the delivery)
    print(f"{item} delivered")
    return item

async def main():
    start = time.time()
    # run all orders concurrently and collect the results
    results = await asyncio.gather(
        order("bread", 2),
        order("milk", 1),
        order("juice", 3),
    )
    print("all delivered:", results, "in %.1fs" % (time.time() - start))

asyncio.run(main())

هر سه سفارش با هم پیش می‌روند، پس کل کار به‌اندازه‌ی طولانی‌ترین سفارش (۳ ثانیه) طول می‌کشد، نه مجموع آن‌ها (۶ ثانیه). اگر به‌جای await asyncio.sleep از time.sleep استفاده می‌کردیم، همه‌چیز بلاک می‌شد و ۶ ثانیه طول می‌کشید — نکته‌ی مهم این است که در asyncio باید از نسخه‌های «غیرمسدودکننده» (async) استفاده کنید.

مثال واقعی: بررسی سلامت ۲۰۰ سرور به‌صورت هم‌زمان

حالا به مسئله‌ی ابتدای درس برگردیم. با asyncio می‌توانیم برای هر سرور یک Task بسازیم و همه را با هم اجرا کنیم؛ چون همه بیشترِ وقتشان را منتظر شبکه‌اند، کل کار به‌جای دویست ثانیه، حدود یک ثانیه طول می‌کشد — و همه در یک ترد:

import asyncio
import time

async def health_check(server_id):
    await asyncio.sleep(1)           # simulate a network request to the server
    return (server_id, "ok")

async def main():
    start = time.time()
    tasks = [asyncio.create_task(health_check(i)) for i in range(200)]
    results = await asyncio.gather(*tasks)
    print(f"checked {len(results)} servers in %.2fs" % (time.time() - start))

asyncio.run(main())   # checked 200 servers in ~1.00s

import asyncio
import time

async def health_check(server_id):
    await asyncio.sleep(1)           # simulate a network request to the server
    return (server_id, "ok")

async def main():
    start = time.time()
    tasks = [asyncio.create_task(health_check(i)) for i in range(200)]
    results = await asyncio.gather(*tasks)
    print(f"checked {len(results)} servers in %.2fs" % (time.time() - start))

asyncio.run(main())   # checked 200 servers in ~1.00s

در دنیای واقعی به‌جای asyncio.sleep، یک درخواست شبکه‌ی غیرمسدودکننده (مثلاً با کتابخانه‌ی aiohttp) قرار می‌گیرد. همین الگو برای خزیدن هزاران صفحه‌ی وب، ارسال انبوه پیام، یا هر کار I/O-boundِ پرتعداد به کار می‌رود و دقیقاً همان جایی است که asyncio می‌درخشد.

جمع‌بندی

• asyncio با یک ترد و بدون بلاک‌کردن پردازنده، هزاران کار I/O را هم‌زمان مدیریت می‌کند؛ جایی که ساختن هزاران ترد ممکن نیست.
• چهار جزء اصلی: هم‌روال (async def)، حلقه‌ی رویداد، Task و await.
• حلقه فقط وقتی هیچ کار آماده‌ای نباشد به خواب می‌رود و با رسیدن رویداد I/O بیدار می‌شود؛ این خواب هیچ CPU مصرف نمی‌کند.
• در asyncio همیشه از نسخه‌های غیرمسدودکننده (مثل asyncio.sleep به‌جای time.sleep) استفاده کنید، وگرنه کل حلقه بلاک می‌شود.