📰 مقالات › 🎓 آموزش › async و await در پایتون — از …

📝 مقاله 🎓 آموزش

async و await در پایتون — از صفر تا درک کامل با مثال‌های اجراشدنی

برنامه‌نویسی ناهمزمان در پایتون را یک بار برای همیشه بفهم — ساده، دقیق، و با کدهایی که خودت می‌توانی اجرا کنی

میثم شبانی

نویسنده

📅 ۱۴۰۵/۰۳/۱۸

⏱ 5 دقیقه

👁 209

❤️ 2

#async-await #asyncio #backend #concurrency #python #taskgroup

📋 فهرست مطالب ▼

اول مفهوم: همزمانی یعنی چه؟
کوروتین (coroutine): قلب async
جادوی واقعی: اجرای همزمان چند کار
گرفتن نتیجه‌ی کارها
پس asyncio.gather چه شد؟
مقایسه‌ی سریع
یک ابزار کاربردی: timeout
مثال واقعی: کجا از این استفاده می‌کنیم؟
جمع‌بندی

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

این مقاله را طوری نوشته‌ایم که اگر تازه‌کاری، مفهوم را کامل بگیری؛ و اگر حرفه‌ای هستی، دقیق‌ترین و مدرن‌ترین روش‌ها (مثل TaskGroup در پایتون ۳.۱۱+) را ببینی. همه‌ی کدها قابل کپی و اجرا هستند — کافی است پایتون ۳.۱۱ یا بالاتر داشته باشی.

⚠️

async کجا به درد می‌خورد؟

async برای کارهای «I/O-bound» عالی است: درخواست شبکه، خواندن از دیتابیس، کار با فایل، صدا زدن API. اما برای کارهای «CPU-bound» (محاسبات سنگین مثل پردازش تصویر) مناسب نیست — برای آن‌ها باید سراغ multiprocessing بروی. این مهم‌ترین نکته‌ای است که باید از همین اول بدانی.

اول مفهوم: همزمانی یعنی چه؟

برنامه‌ی معمولی (synchronous) خط‌به‌خط اجرا می‌شود: تا یک خط تمام نشود، خط بعدی شروع نمی‌شود. اگر خطی منتظر دانلود یک فایل باشد، کل برنامه همان‌جا متوقف می‌ماند تا دانلود تمام شود.

برنامه‌ی ناهمزمان (asynchronous) فرق دارد: وقتی به یک نقطه‌ی «انتظار» می‌رسد، کنترل را به کارهای دیگر می‌دهد و وقتی انتظار تمام شد، برمی‌گردد. توجه کن: این یعنی «همزمانی» (concurrency) نه «موازی‌سازی» (parallelism). همه چیز در یک thread اجرا می‌شود — فقط هوشمندانه بین کارها جابه‌جا می‌شویم.

کوروتین (coroutine): قلب async

هر تابعی که با async def تعریف شود، یک «کوروتین» است. نکته‌ی کلیدی که خیلی‌ها را گیج می‌کند: وقتی یک کوروتین را صدا می‌زنی، اجرا نمی‌شود! فقط یک «شیء کوروتین» ساخته می‌شود. برای اجرای واقعی باید آن را await کنی یا با asyncio.run() اجرا کنی.

🐍 PYTHON — اولین کوروتین — make_coffee

import asyncio


# با async def یک کوروتین می‌سازیم
async def make_coffee():
    # شبیه‌سازی زمان دم‌کشیدن قهوه (۲ ثانیه)
    await asyncio.sleep(2)
    return "قهوه آماده است"


async def main():
    # توجه: صدا زدن کوروتین، آن را اجرا نمی‌کند
    result = await make_coffee()  # اینجا منتظر می‌مانیم
    print(result)


# asyncio.run نقطه‌ی شروع برنامه‌ی async است
asyncio.run(main())

# خروجی (بعد از ۲ ثانیه):
# قهوه آماده است

🔴

اشتباه شماره یک: فراموش کردن await

اگر یک کوروتین را بدون await صدا بزنی (مثلاً فقط make_coffee())، پایتون هشدار "coroutine was never awaited" می‌دهد و کد داخلش اصلاً اجرا نمی‌شود. این رایج‌ترین اشتباه تازه‌کارهاست. قانون طلایی: هر کوروتین یا باید await شود یا داخل asyncio.run() اجرا شود.

جادوی واقعی: اجرای همزمان چند کار

تا اینجا فقط یک کار را await کردیم. قدرت async وقتی معلوم می‌شود که چند کار را همزمان اجرا کنیم. اول بیایید ببینیم اگر کارها را پشت‌سرهم (sequential) اجرا کنیم چه می‌شود:

🐍 PYTHON — روش کند — اجرای پشت‌سرهم (۳ ثانیه)

import asyncio
import time


async def say_after(delay, what):
    # هر کار به اندازه‌ی delay ثانیه طول می‌کشد
    await asyncio.sleep(delay)
    print(what)


async def main():
    print(f"شروع: {time.strftime('%X')}")

    # پشت سر هم — هر کدام منتظر قبلی می‌ماند
    await say_after(1, "سلام")
    await say_after(2, "دنیا")

    print(f"پایان: {time.strftime('%X')}")


asyncio.run(main())

# خروجی — مجموعاً حدود ۳ ثانیه طول می‌کشد:
# شروع: 12:00:00
# سلام      (بعد از ۱ ثانیه)
# دنیا      (بعد از ۲ ثانیه دیگر)
# پایان: 12:00:03

سه ثانیه طول کشید چون کار دوم تا تمام شدن کار اول صبر کرد. حالا بیایید همان دو کار را «همزمان» اجرا کنیم تا تفاوت را ببینی. از پایتون ۳.۱۱ به بعد، بهترین و امن‌ترین راه، استفاده از asyncio.TaskGroup است که «همزمانی ساختارمند» را پیاده می‌کند.

🐍 PYTHON — روش مدرن و سریع — TaskGroup (۲ ثانیه)

import asyncio
import time


async def say_after(delay, what):
    await asyncio.sleep(delay)
    print(what)


async def main():
    print(f"شروع: {time.strftime('%X')}")

    # هر دو کار با هم شروع می‌شوند
    async with asyncio.TaskGroup() as tg:
        tg.create_task(say_after(1, "سلام"))
        tg.create_task(say_after(2, "دنیا"))
    # وقتی از بلاک with خارج می‌شویم، هر دو کار تمام شده‌اند

    print(f"پایان: {time.strftime('%X')}")


asyncio.run(main())

# خروجی — فقط حدود ۲ ثانیه (نه ۳ ثانیه):
# شروع: 12:00:00
# سلام      (بعد از ۱ ثانیه)
# دنیا      (بعد از ۲ ثانیه — همزمان با بالایی شمرده شد)
# پایان: 12:00:02

✅

تفاوت را دیدی؟

چرا ۲ ثانیه و نه ۳ ثانیه؟ چون هر دو کار در لحظه‌ی صفر شروع شدند. کار اول بعد از ۱ ثانیه و کار دوم بعد از ۲ ثانیه (که همزمان شمرده شد) تمام شد. این همان صرفه‌جویی در زمان است که async به ما می‌دهد — مخصوصاً وقتی ده‌ها درخواست شبکه را همزمان بفرستی.

گرفتن نتیجه‌ی کارها

وقتی با TaskGroup کار می‌کنی، برای گرفتن نتیجه‌ی هر Task باید بعد از بسته شدن بلاک with، متد result() را صدا بزنی. این یک نکته‌ی مهم است که حتی حرفه‌ای‌ها گاهی فراموش می‌کنند.

🐍 PYTHON — گرفتن نتیجه از چند Task همزمان

import asyncio


async def fetch_user(user_id):
    # شبیه‌سازی گرفتن اطلاعات کاربر از دیتابیس
    await asyncio.sleep(1)
    return {"id": user_id, "name": f"User {user_id}"}


async def main():
    async with asyncio.TaskGroup() as tg:
        task_1 = tg.create_task(fetch_user(1))
        task_2 = tg.create_task(fetch_user(2))
        task_3 = tg.create_task(fetch_user(3))

    # نتیجه را بعد از بسته شدن بلاک می‌گیریم
    users = [task_1.result(), task_2.result(), task_3.result()]
    for user in users:
        print(user)


asyncio.run(main())

# خروجی (بعد از حدود ۱ ثانیه — نه ۳ ثانیه):
# {'id': 1, 'name': 'User 1'}
# {'id': 2, 'name': 'User 2'}
# {'id': 3, 'name': 'User 3'}

پس asyncio.gather چه شد؟

شاید در کدهای قدیمی‌تر زیاد asyncio.gather() را دیده باشی. این روش هنوز کار می‌کند و برای کارهای ساده خوب است، اما یک تفاوت مهم با TaskGroup دارد که باید بدانی.

🐍 PYTHON — روش قدیمی‌تر — asyncio.gather

import asyncio


async def fetch_user(user_id):
    await asyncio.sleep(1)
    return {"id": user_id, "name": f"User {user_id}"}


async def main():
    # gather نتیجه‌ها را به‌ترتیب در یک لیست برمی‌گرداند
    users = await asyncio.gather(
        fetch_user(1),
        fetch_user(2),
        fetch_user(3),
    )
    print(users)


asyncio.run(main())

# خروجی (بعد از حدود ۱ ثانیه):
# [{'id': 1, 'name': 'User 1'}, {'id': 2, 'name': 'User 2'}, {'id': 3, 'name': 'User 3'}]

💫

TaskGroup یا gather؟

تفاوت کلیدی: اگر یکی از کارها در gather خطا بدهد، بقیه‌ی کارها کنسل نمی‌شوند و خطا ممکن است گم شود (مگر اینکه return_exceptions=True بگذاری و خودت نتیجه‌ها را چک کنی). اما در TaskGroup، اگر یک کار خطا بدهد، بقیه خودکار کنسل می‌شوند و خطاها در قالب ExceptionGroup بالا می‌آیند. برای همین در پایتون ۳.۱۱+ توصیه می‌شود برای کارهای مرتبط، TaskGroup را ترجیح بدهی.

مقایسه‌ی سریع

ویژگی	asyncio.gather	asyncio.TaskGroup
نسخه پایتون	همه نسخه‌ها	۳.۱۱ به بعد
اگر یک کار خطا دهد	بقیه ادامه می‌دهند	بقیه کنسل می‌شوند
مدیریت خطا	دستی (return_exceptions)	خودکار (ExceptionGroup)
گرفتن نتیجه	لیست مستقیم	با task.result()
توصیه‌ی امروز	کارهای ساده	روش پیش‌فرض

یک ابزار کاربردی: timeout

گاهی نمی‌خواهی بی‌نهایت منتظر یک کار بمانی. مثلاً اگر یک API بعد از ۳ ثانیه جواب نداد، می‌خواهی بی‌خیال شوی. برای این کار asyncio.wait_for() عالی است:

🐍 PYTHON — محدود کردن زمان انتظار با wait_for

import asyncio


async def slow_api():
    # این API عمداً کند است (۵ ثانیه)
    await asyncio.sleep(5)
    return "پاسخ API"


async def main():
    try:
        # حداکثر ۳ ثانیه صبر کن
        result = await asyncio.wait_for(slow_api(), timeout=3)
        print(result)
    except asyncio.TimeoutError:
        print("API دیر جواب داد — بی‌خیال شدیم")


asyncio.run(main())

# خروجی (بعد از ۳ ثانیه):
# API دیر جواب داد — بی‌خیال شدیم

مثال واقعی: کجا از این استفاده می‌کنیم؟

async در دنیای واقعی همه‌جا هست. فریم‌ورک FastAPI کاملاً بر پایه‌ی async ساخته شده. کتابخانه‌هایی مثل httpx و aiohttp برای درخواست‌های شبکه‌ی همزمان از آن استفاده می‌کنند. حتی وقتی با صف پیام (مثل Redis یا Kafka) کار می‌کنی، الگوهای async به‌شدت پرکاربردند.

💡

ادامه‌ی مسیر

اگر می‌خواهی ببینی async در یک فریم‌ورک واقعی چطور به کار می‌رود، مقاله‌های «جنگو ۶» و «Redis، RabbitMQ یا Kafka؟» در همین سایت نقطه‌ی شروع خوبی هستند — مخصوصاً برای دیدن اینکه کارهای I/O-bound در عمل چطور مدیریت می‌شوند.

جمع‌بندی

async def یک کوروتین می‌سازد؛ صدا زدنش آن را اجرا نمی‌کند — باید await شود
await یعنی «منتظر بمان ولی thread را آزاد کن تا کار دیگری انجام شود»
asyncio.run(main()) نقطه‌ی شروع برنامه‌ی async است
برای اجرای همزمان چند کار، در پایتون ۳.۱۱+ از asyncio.TaskGroup استفاده کن
نتیجه‌ی Taskها را با task.result() بعد از بلاک with بگیر
async برای I/O-bound عالی است، برای CPU-bound مناسب نیست
برای محدود کردن زمان انتظار از asyncio.wait_for() استفاده کن

حالا مفهوم async و await را کامل می‌دانی — هم تئوری، هم کد اجراشدنی. بهترین راه برای تثبیت یادگیری این است که همین کدها را کپی کنی، اجرا کنی، و کمی تغییرشان بدهی. مثلاً delayها را عوض کن و ببین خروجی چطور تغییر می‌کند. موفق باشی!

🤍 لایک