اگر تا به حال کدی نوشته‌ای که باید هم‌زمان چند درخواست شبکه بفرستد، چند فایل را دانلود کند یا با چند API صحبت کند و دیده‌ای که برنامه‌ات روی هر درخواست چند ثانیه قفل می‌شود، دقیقاً جای درستی آمده‌ای. در پایتون راه‌حل تمیز این مشکل AsyncIO است؛ مدلی که به تو اجازه می‌دهد بدون شلوغی Threadها، هزاران عملیات I/O را به صورت هم‌زمان و با یک Event Loop مدیریت کنی. در این آموزش قرار است از صفر تا اجرای واقعی asyncio را به زبان ساده و با مثال‌های قابل اجرا یاد بگیری.

پاسخ کوتاه: AsyncIO چیست و چرا باید یاد بگیریم؟

AsyncIO کتابخانه‌ی استاندارد پایتون برای برنامه‌نویسی ناهمگام (asynchronous) است که با دو کلمه‌ی کلیدی async و await کار می‌کند و تمام تسک‌های منتظر I/O را روی یک Event Loop واحد می‌چرخاند. نتیجه؟ به جای منتظر ماندن ترتیبی برای هر درخواست، همه‌ی درخواست‌ها تقریباً هم‌زمان جلو می‌روند و سرعت برنامه‌های I/O-bound تا چند برابر بیشتر می‌شود. در ادامه قدم‌به‌قدم پیاده‌سازی آن را می‌بینی.

AsyncIO چیست و چه تفاوتی با Thread دارد؟

خیلی‌ها AsyncIO را با Thread اشتباه می‌گیرند. هر دو «هم‌زمانی» می‌آورند، اما سازوکارشان زمین تا آسمان فرق دارد. Threadها واقعاً نخ‌های جداگانه‌ی سیستم‌عامل هستند؛ هر کدام حافظه و سربار خودش را دارد و سوییچ بین آن‌ها هزینه‌بر است. AsyncIO اما تک‌نخی و Cooperative است؛ یعنی خودت با await می‌گویی «اینجا منتظر I/O هستم، برو سراغ تسک بعدی». این مدل برای کارهایی که بیشتر زمان‌شان صرف انتظار می‌شود (درخواست HTTP، کوئری دیتابیس، خواندن فایل، بات تلگرام) فوق‌العاده سبک و سریع است.

Threading: مناسب کارهای CPU-bound اگر از GIL عبور کنی یا از سیستم‌عامل کمک بگیری، اما سربار بالا و دیباگ سخت‌تر.
AsyncIO: عالی برای I/O-bound، هزاران کانکشن با مصرف رم کم، کد خواناتر با async/await.
Multiprocessing: برای کارهای سنگین CPU که واقعاً باید روی چند هسته تقسیم شوند.

اگر پروژه‌ات مثلاً یک API با FastAPI، یک کراولر یا یک ربات تلگرام است، AsyncIO انتخاب اول توست. اتفاقاً اگر قبلاً محیط پایتون را با مدیریت‌کننده‌های مدرن ساخته باشی، کارت راحت‌تر هم می‌شود؛ برای همین پیشنهاد می‌کنم نگاهی به آموزش uv برای مدیریت محیط و پکیج‌های پایتون بیندازی تا محیط مجازی و وابستگی‌ها را تمیز نگه داری.

مفهوم رویداد لوپ (Event Loop) به زبان ساده

قلب AsyncIO یک حلقه‌ی بی‌پایان به نام Event Loop است. لوپ لیست تسک‌های آماده را نگه می‌دارد، یکی را اجرا می‌کند، به await می‌رسد، آن تسک را «می‌خواباند» و سراغ تسک بعدی می‌رود. وقتی I/O تمام شد، لوپ دوباره تسک را بیدار می‌کند. خودت لازم نیست لوپ را دستی بچرخانی؛ از پایتون 3.7 به بعد کافی است asyncio.run(main()) را صدا بزنی.

نکته‌ی طلایی: هر تابعی که با async def تعریف می‌شود یک کوروتین است و تا await نشود اجرا نمی‌شود. فراموش کردن await شایع‌ترین باگ تازه‌کارهاست و هیچ خطایی هم نمی‌دهد، فقط تسک اجرا نمی‌شود!

چه زمانی از AsyncIO استفاده کنیم و چه زمانی نه؟

همه‌ی کدها به AsyncIO نیاز ندارند. انتخاب درست، تفاوت بین «سریع‌تر شدن» و «پیچیده‌تر شدن بی‌دلیل» است.

کارهای I/O-bound در برابر CPU-bound

I/O-bound: برنامه بیشتر منتظر شبکه، دیسک یا دیتابیس است. مثال‌ها: درخواست به APIها، دانلود فایل، ارسال ایمیل، چت‌بات. اینجا AsyncIO معجزه می‌کند چون زمان انتظار را با کار مفید پر می‌کند.

CPU-bound: برنامه بیشتر محاسبه می‌کند؛ مثل پردازش تصویر، آموزش مدل، محاسبات ریاضی سنگین. اینجا GIL پایتون گلوگاه می‌شود و AsyncIO کمکی نمی‌کند؛ باید سراغ Multiprocessing یا کتابخانه‌های آزاد از GIL (مثل NumPy) بروی.

اگر 80٪ زمان برنامه انتظار شبکه است → AsyncIO
اگر 80٪ زمان برنامه محاسبه روی CPU است → Multiprocessing
اگر ترکیبی است → AsyncIO برای I/O + ProcessPool برای CPU

اولین کد AsyncIO؛ از async و await تا اجرا

بیایید اولین کد ناهمگام را بنویسیم. فرض کن می‌خواهی دو کار که هر کدام 2 ثانیه طول می‌کشد را اجرا کنی. در حالت عادی 4 ثانیه طول می‌کشد، اما با AsyncIO حدود 2 ثانیه.

import asyncio
import time

async def task(name, delay):
    print(f"شروع {name}")
    await asyncio.sleep(delay)  # شبیه‌سازی I/O مثل درخواست شبکه
    print(f"پایان {name}")
    return f"{name} تمام شد"

async def main():
    start = time.time()
    # اجرای ترتیبی (کند)
    # await task("A", 2)
    # await task("B", 2)

    # اجرای هم‌زمان (سریع)
    results = await asyncio.gather(
        task("A", 2),
        task("B", 2)
    )
    print(results)
    print(f"زمان کل: {time.time() - start:.1f} ثانیه")

asyncio.run(main())

خروجی می‌بینی که هر دو تسک با هم شروع می‌شوند و کل زمان حدود 2 ثانیه است نه 4 ثانیه. رازش همان await asyncio.sleep است که کنترل را به لوپ برمی‌گرداند.

ساخت یک تابع async ساده و نکات حیاتی

هر تابع async def را فقط با await یا با asyncio.create_task می‌توان اجرا کرد.
داخل تابع async نمی‌توانی مستقیم time.sleep بگذاری؛ باید از await asyncio.sleep استفاده کنی، وگرنه کل لوپ قفل می‌شود.
برای صدا زدن کد sync داخل async از await asyncio.to_thread(func) (پایتون 3.9+) استفاده کن.

برای درخواست‌های HTTP واقعی به جای sleep از کتابخانه‌های async مثل httpx یا aiohttp استفاده می‌کنی؛ مثلاً await httpx.AsyncClient().get(url).

مدیریت چند تسک با gather و create_task

وقتی تعداد تسک‌ها زیاد شد، دو ابزار اصلی داری: gather و create_task.

اجرای هم‌زمان با asyncio.gather

gather ساده‌ترین راه برای «همه را با هم اجرا کن و منتظر همه بمان» است. اگر یکی خطا بدهد، بقیه هم لغو می‌شوند مگر return_exceptions=True بگذاری. این گزینه برای کراولرها عالی است چون می‌خواهی حتی اگر یکی از URLها خطا داد، بقیه نتایج را داشته باشی.

results = await asyncio.gather(
    fetch("https://api1.example.com"),
    fetch("https://api2.example.com"),
    fetch("https://api3.example.com"),
    return_exceptions=True
)

کنترل تسک‌ها با create_task و cancel

وقتی می‌خواهی تسکی را در پس‌زمینه روشن نگه داری، به آن اسم بدهی یا وسط راه لغو کنی، create_task به کارت می‌آید.

task = asyncio.create_task(fetch_large_file())
# ... کارهای دیگر
if need_cancel:
    task.cancel()
    try:
        await task
    except asyncio.CancelledError:
        print("تسک لغو شد")

یادت باشد تسک‌های ساخته شده با create_task اگر await نشوند، در پس‌زمینه رها می‌مانند و ممکن است خطایشان گم شود. همیشه یا await کن یا به یک لیست اضافه و در پایان gather کن.

کار با Semaphore و Queue برای کنترل هم‌زمانی

اجرای 1000 درخواست هم‌زمان شاید سرور مقصد را اذیت کند یا رم را پر کند. برای همین باید هم‌زمانی را محدود کنی. Semaphore مثل نگهبان درِ ورودی عمل می‌کند.

sem = asyncio.Semaphore(5)  # فقط 5 تسک هم‌زمان

async def limited_fetch(url):
    async with sem:
        return await fetch(url)

await asyncio.gather(*[limited_fetch(url) for url in urls])

برای الگوی تولیدکننده-مصرف‌کننده، asyncio.Queue عالی است. یک کوروتین URLها را در صف می‌گذارد و چند مصرف‌کننده آن‌ها را برمی‌دارند. این الگو در خزنده‌های وب و صف ارسال پیام بسیار کاربردی است و کد را تمیز و قابل تست نگه می‌دارد.

Semaphore: محدود کردن تعداد تسک‌های فعال (مثلاً 10 درخواست هم‌زمان به API).
Queue: توزیع کار بین چند worker ناهمگام.
Lock: وقتی چند تسک می‌خواهند به یک منبع مشترک (مثل فایل) بنویسند.

خطاهای رایج AsyncIO و نحوه دیباگ آن‌ها

سه خطای کلاسیک که هر تازه‌کاری با آن گیر می‌کند:

فراموش کردن await: تابع کوروتین ساخته می‌شود اما اجرا نمی‌شود. پایتون هشدار coroutine was never awaited می‌دهد. همیشه نتیجه‌ی تابع async را await کن.
استفاده از کتابخانه‌ی sync داخل async: مثلاً requests.get داخل تابع async کل لوپ را بلاک می‌کند. جایگزین async مثل httpx یا اجرای آن در to_thread را به کار ببر.
RuntimeWarning: coroutine was never awaited + قفل شدن لوپ: معمولاً به خاطر time.sleep یا عملیات سنگین CPU داخل لوپ است. آن را به thread یا process جدا بفرست.

برای دیباگ، پایتون حالت PYTHONASYNCIODEBUG=1 و asyncio.run(..., debug=True) دارد که تسک‌های کند را گزارش می‌دهد. همچنین لاگ‌گیری ساختاریافته کمک زیادی می‌کند؛ اگر پروژه‌ات بزرگ شد، بد نیست پایپ‌لاین تست و انتشار خودکار هم داشته باشی؛ یک نمونه‌ی کامل را در آموزش ساخت پایپ‌لاین CI/CD با GitHub Actions می‌بینی که چطور روی هر push تست‌های async را اجرا کنی.

مقایسه AsyncIO با Threading و Multiprocessing

برای تصمیم سریع، این جدول را ذخیره کن:

ویژگی	AsyncIO	Threading	Multiprocessing
مدل اجرا	تک‌نخ، cooperative	چند نخ سیستم‌عامل	چند پردازه جدا
بهترین کاربرد	I/O-bound (شبکه، API)	I/O-bound سبک + کتابخانه sync	CPU-bound سنگین
مصرف حافظه	بسیار کم (هزاران تسک)	متوسط (هر Thread ~8MB)	زیاد (هر Process حافظه جدا)
پیچیدگی دیباگ	متوسط (نیاز به await)	سخت (Race Condition)	سخت (IPC)
سازگاری با GIL	گیر نمی‌کند چون I/O آزاد می‌کند	گیر می‌کند روی CPU	دور می‌زند (هر پردازه GIL جدا)
مثال کتابخانه	httpx, aiohttp, asyncpg	requests, threading	concurrent.futures.ProcessPool

جمع‌بندی: اگر کارت درخواست‌های زیاد شبکه است، AsyncIO هم سریع‌تر است هم تمیزتر. اگر محاسبات سنگین داری، Multiprocessing را انتخاب کن و اگر مجبوری با کتابخانه‌ی قدیمی sync کار کنی، آن را داخل to_thread بگذار تا لوپ قفل نشود.

پروژه کوچک کاربردی؛ دانلود هم‌زمان چند فایل با httpx

بیایید یک اسکریپت واقعی بسازیم که 5 URL را هم‌زمان دانلود کند و سرعت را با Semaphore کنترل کند. این الگو را می‌توانی برای بک‌آپ APIها، دانلود تصاویر یا کراول قیمت‌ها استفاده کنی.

import asyncio, httpx, time

URLS = [
    "https://jsonplaceholder.typicode.com/posts/1",
    "https://jsonplaceholder.typicode.com/posts/2",
    "https://jsonplaceholder.typicode.com/posts/3",
    "https://jsonplaceholder.typicode.com/posts/4",
    "https://jsonplaceholder.typicode.com/posts/5",
]

sem = asyncio.Semaphore(3)

async def fetch(client, url):
    async with sem:
        r = await client.get(url, timeout=10)
        r.raise_for_status()
        print(f"✓ {url} -> {len(r.text)} بایت")
        return r.json()

async def main():
    start = time.time()
    async with httpx.AsyncClient() as client:
        results = await asyncio.gather(*[fetch(client, u) for u in URLS])
    print(f"تمام شد در {time.time()-start:.2f} ثانیه | {len(results)} آیتم")

if __name__ == "__main__":
    asyncio.run(main())

این کد را با pip install httpx اجرا کن. برای تست سرعت، همین URLها را یک بار با requests ترتیبی هم امتحان کن؛ اختلاف زمان ملموس است. اگر خواستی همین اسکریپت هر روز خودکار اجرا شود، همان پایپ‌لاین GitHub Actions که بالاتر لینک شد به کارت می‌آید.

نکته‌ی حرفه‌ای: برای ذخیره نتایج در دیتابیس async از asyncpg (پستگرس) یا aiosqlite استفاده کن تا کل مسیر از درخواست تا ذخیره، ناهمگام بماند.

سوالات متداول

آیا AsyncIO سرعت CPU را هم بیشتر می‌کند؟

خیر. AsyncIO برای کارهای I/O-bound طراحی شده و روی محاسبات سنگین CPU تأثیری ندارد، چون GIL همچنان یک نخ را در یک لحظه اجرا می‌کند. برای CPU-bound از Multiprocessing یا کتابخانه‌های رها از GIL استفاده کن.

تفاوت asyncio.gather و asyncio.wait چیست؟

gather نتایج را به ترتیب ورودی برمی‌گرداند و کار با آن ساده‌تر است، اما اگر یکی خطا دهد همه لغو می‌شوند (مگر return_exceptions). wait کنترل بیشتری می‌دهد؛ می‌توانی بگویی با اولین موفقیت برگرد (FIRST_COMPLETED) یا منتظر همه بمان (ALL_COMPLETED) و تسک‌های ناتمام را خودت مدیریت کنی.

آیا می‌توانم کد sync قدیمی را بدون بازنویسی در AsyncIO استفاده کنم؟

بله. تابع sync را با await asyncio.to_thread(my_sync_func, arg) در Thread جدا اجرا کن تا Event Loop بلاک نشود. این راه‌حل برای کتابخانه‌هایی مثل requests یا پردازش فایل‌های بزرگ عالی است، اما اگر تعدادشان زیاد شد بهتر است به نسخه‌ی async آن کتابخانه مهاجرت کنی.

آموزش AsyncIO در پایتون؛ اجرای هم‌زمان تسک‌ها به زبان ساده