کتابخانه asyncio در پایتون — راهنمای جامع

asyncio یکی از مهم‌ترین کتابخانه‌های استاندارد پایتون برای برنامه‌نویسی ناهمگام (asynchronous) است. این کتابخانه به شما امکان می‌دهد تا کارهای وابسته به I/O را به صورت هم‌زمان و کارآمد اجرا کنید، بدون اینکه از چندنخی (threading) یا فرآیندهای جداگانه استفاده کنید. در این مقاله با مفاهیم پایه، الگوهای کاربردی، مثال‌های واقعی و نکات پیشرفته آشنا می‌شوید.

چرا asyncio؟ (مزایا و موارد استفاده)

• مناسب برای I/O-bound: شبکه، فایل‌ها، پایگاه‌داده‌های async، و وب‌سرویس‌ها.
• کارایی بهتر در مقایسه با thread برای هزاران اتصال هم‌زمان، به شرطی که عملیات‌های cpu-heavy را به executor منتقل کنید.
• ساختار واضح با async/await که خوانایی کد را افزایش می‌دهد.

مفاهیم کلیدی

• Coroutine: تابعی که با async تعریف می‌شود و قابلیت معلق شدن دارد.
• Event loop: موتور اجرایی که کرروتین‌ها را زمان‌بندی می‌کند.
• Task: کرروتینی که به event loop واگذار شده و قابل مدیریت است.
• Future: نمایشگر نتیجهٔ آتی یک محاسبه.

توابع و APIهای پرکاربرد

مثال پایه: کرروتین‌ها و run

import asyncio

async def say(name, delay):
    await asyncio.sleep(delay)
    return f"Hello {name} after {delay}"

async def main():
    results = await asyncio.gather(
        say("Alice", 1),
        say("Bob", 2),
        say("Carol", 0.5)
    )
    print(results)

asyncio.run(main())

در این مثال سه کرروتین هم‌زمان اجرا می‌شوند و با asyncio.gather منتظر بازگشت همهٔ نتایج می‌مانیم. asyncio.sleep به‌عنوان یک عملیات I/O غیرفعال شبیه‌سازی شده عمل می‌کند، بنابراین event loop می‌تواند در زمان خواب سایر کرروتین‌ها کارهای دیگر انجام دهد.

مدیریت تعداد هم‌زمان اتصال‌ها (Semaphore)

import asyncio
import aiohttp

sem = asyncio.Semaphore(10)

async def fetch(session, url):
    async with sem:
        async with session.get(url) as resp:
            return await resp.text()

async def main(urls):
    async with aiohttp.ClientSession() as session:
        tasks = [asyncio.create_task(fetch(session, u)) for u in urls]
        return await asyncio.gather(*tasks)

در این نمونه از aiohttp برای درخواست‌های HTTP استفاده شده و با Semaphore تعداد هم‌زمانی که اجازهٔ ارسال درخواست دارند محدود می‌شود. این الگو برای جلوگیری از فشار بیش‌از‌حد روی سرور یا مصرف زیاد منابع مفید است.

بهبود: اضافه کردن timeout و مدیریت خطا

import asyncio
import aiohttp

async def fetch_with_timeout(session, url, timeout=10):
    try:
        return await asyncio.wait_for(session.get(url), timeout=timeout)
    except asyncio.TimeoutError:
        return None

در این کد، wait_for باعث می‌شود اگر درخواست طولانی شد، لغو شود و برنامه پاسخ‌پذیری خود را حفظ کند. از این روش برای جلوگیری از بلوکه شدن منابع استفاده می‌شود.

کارهای CPU-bound — استفاده از run_in_executor

import asyncio
import concurrent.futures

def cpu_heavy(x):
    # محاسبه‌ای سنگین که نباید در event loop اجرا شود
    return sum(i*i for i in range(x))

async def main():
    loop = asyncio.get_running_loop()
    with concurrent.futures.ProcessPoolExecutor() as pool:
        result = await loop.run_in_executor(pool, cpu_heavy, 10_000_000)
        print(result)

asyncio.run(main())

برای کارهای سنگین محاسباتی بهتر است از ProcessPoolExecutor یا ThreadPoolExecutor استفاده کنید تا event loop مسدود نشود. run_in_executor تابع را در یک نخ یا فرآیند جدا اجرا کرده و نتیجه را به صورت ناهمگام بازمی‌گرداند.

لغو (Cancellation) و نکات ایمن

• Taskها قابل لغو هستند؛ با task.cancel() می‌توانید درخواست لغو بدهید.
• هنگام لغو، حتماً منابع (مانند sessionها) را در بلوک‌های try/finally یا async with آزاد کنید.

نکات عملی و توصیه‌های حرفه‌ای

• برای I/O-bound، asyncio معمولاً بهترین انتخاب است؛ برای CPU-bound از executor استفاده کنید.
• از asyncio.run به عنوان نقطهٔ ورود استفاده کنید تا مدیریت event loop ساده شود.
• برای برنامه‌های بزرگ از ساختارهای سطح بالاتر مانند asyncio.Queue، Semaphore و Barrier بهره ببرید.
• خطاها را در gather با return_exceptions=True مدیریت کنید یا از try/except در کرروتین‌ها استفاده کنید.
• آزمایش عملکرد با ابزارهایی مثل aiohttp و پروفایلینگ نیمه‌بلادرنگ (sampling) انجام دهید.

مثال کامل: دانلود چند فایل با کنترل هم‌زمانی و timeout

import asyncio
import aiohttp

async def download(url, session, sem, timeout=10):
    try:
        async with sem:
            async with session.get(url, timeout=timeout) as resp:
                return await resp.read()
    except Exception as e:
        return f"Error: {e}"

async def download_all(urls):
    sem = asyncio.Semaphore(5)
    async with aiohttp.ClientSession() as session:
        tasks = [asyncio.create_task(download(u, session, sem)) for u in urls]
        return await asyncio.gather(*tasks, return_exceptions=True)

# asyncio.run(download_all(list_of_urls))

این الگو ترکیبی از Semaphore، aiohttp، timeout و gather است. return_exceptions=True کمک می‌کند تا خطاها به عنوان نتیجه برگردانده شوند و یک Task خراب باعث لغو بقیه نشود. در عمل می‌توانید خطاها را لاگ کنید یا تلاش مجدد (retry) پیاده کنید.

خلاصه و جمع‌بندی

asyncio ابزار قدرتمندی برای ساخت برنامه‌های ناهمگام در پایتون فراهم می‌کند؛ مخصوصاً برای سرویس‌های شبکه‌ای با تعداد زیاد اتصال هم‌زمان. آشنایی با event loop، کرروتین‌ها، Taskها و الگوهای مدیریتی (Semaphore، Queue، Executor) به شما کمک می‌کند تا برنامه‌هایی مقیاس‌پذیر و پاسخگو بنویسید. در نهایت، همیشه توجه کنید که عملیات CPU-bound را از event loop جدا کنید و برای مدیریت خطا، timeout و لغو تمهیدات لازم را در نظر بگیرید.

لطفاً از کمبود ها و مشکلات این محتوا برای ما بنویسید

ارسال