آموزش ماژول concurrent.futures در پایتون (همزمانی مدرن)

سطح مقدماتی (Beginner Level)

ماژول concurrent.futures یکی از ابزارهای قدرتمند و مدرن در کتابخانه استاندارد پایتون است که در نسخه 3.2 معرفی شد. این ماژول یک رابط سطح بالا (High-level Interface) برای اجرای وظایف به صورت ناهمگام (Asynchronous) با استفاده از Threadها یا Processها فراهم می‌کند.

برخلاف ماژول‌های قدیمی‌تر threading و multiprocessing که نیاز به مدیریت دستی قفل‌ها و صف‌ها دارند، concurrent.futures کار را با استفاده از Executorها بسیار ساده کرده است.

مفاهیم پایه: Executor چیست؟

کلاس پایه Executor دو زیرمجموعه اصلی دارد که باید تفاوت آن‌ها را بدانید:

1. ThreadPoolExecutor: برای کارهایی که I/O Bound هستند (مثل دانلود فایل، درخواست وب، یا خواندن از دیتابیس). در این حالت، زمان زیادی صرف انتظار می‌شود.
2. ProcessPoolExecutor: برای کارهایی که CPU Bound هستند (مثل محاسبات ریاضی سنگین، پردازش تصویر). در این حالت، پردازنده درگیر محاسبات است.

استفاده از ThreadPoolExecutor

ساده‌ترین راه برای استفاده از این ماژول، استفاده از Context Manager (with) است که مدیریت منابع و بستن Threadها را خودکار انجام می‌دهد.

مثال ۱: تعریف تابع (استاتیک)

ابتدا تابعی را تصور کنید که شبیه‌سازی یک عملیات زمان‌بر (مثل دانلود) است.

# این کد به تنهایی اجرا نمی‌شود و فقط تعریف تابع است
import time

def worker(name):
    print(f"شروع کار {name}...")
    time.sleep(2)  # شبیه‌سازی انتظار
    return f"پایان کار {name}"

مثال ۲: اجرای ساده با submit (تعاملی)

در این مثال، یک وظیفه تکی را با متد submit به استخر (Pool) ارسال می‌کنیم.

مثال ۳: اجرای گروهی با map (تعاملی)

متد map بسیار شبیه به تابع map استاندارد پایتون است، با این تفاوت که وظایف را به صورت همزمان اجرا می‌کند.

استفاده از ProcessPoolExecutor

زمانی که محاسبات سنگین دارید، استفاده از Threadها به دلیل وجود GIL (قفل مفسر جهانی پایتون) کارساز نیست و باید از Processها استفاده کنید تا از تمام هسته‌های CPU بهره ببرید.

مثال ۱: محاسبات سنگین (استاتیک)

این یک تابع محاسباتی است که فقط CPU را درگیر می‌کند.

def cpu_bound_task(number):
    count = 0
    for i in range(number):
        count += i
    return count

مثال ۲: اجرای موازی پردازش‌ها (تعاملی)

نکته: در ویندوز و برخی محیط‌ها، کدهای multiprocessing حتماً باید داخل بلاک if __name__ == '__main__': باشند.

سطح پیشرفته (Professional Level)

در سطح حرفه‌ای، ما نیاز به کنترل دقیق‌تری روی وظایف داریم. متدهایی مثل submit شی‌ای به نام Future برمی‌گردانند که نماینده‌ای برای نتیجه‌ای است که در آینده آماده خواهد شد. مدیریت خطاها، کنسل کردن وظایف و پردازش نتایج به محض آماده شدن (نه به ترتیب ورودی) از مباحث مهم این بخش است.

شیء Future و متدهای آن

شیء Future متدهای مهمی دارد:

• result(timeout=None): نتیجه را برمی‌گرداند (در صورت نیاز صبر می‌کند).
• done(): بررسی می‌کند آیا کار تمام شده است یا خیر.
• cancel(): تلاش می‌کند وظیفه را قبل از شروع لغو کند.
• add_done_callback(fn): وقتی کار تمام شد، تابع fn را اجرا می‌کند.

مثال ۱: بررسی وضعیت Future (تعاملی)

پردازش نتایج با as_completed

تابع map نتایج را به همان ترتیبی که ورودی داده‌اید برمی‌گرداند. اما گاهی می‌خواهید هر کدام که زودتر تمام شد را پردازش کنید. برای این کار از as_completed استفاده می‌کنیم.

مثال ۱: ساختار کد (استاتیک)

from concurrent.futures import as_completed

# لیستی از Futureها ایجاد می‌کنیم
futures = [executor.submit(task, arg) for arg in args]

# هر کدام که زودتر تمام شود وارد حلقه می‌شود
for future in as_completed(futures):
    print(future.result())

مثال ۲: سناریوی واقعی با زمان‌های مختلف (تعاملی)

در این مثال سه وظیفه با زمان‌های متفاوت اجرا می‌شوند. مشاهده کنید که ترتیب خروجی بر اساس زمان اتمام است، نه ترتیب ارسال.

مدیریت استثناها (Exception Handling)

در concurrent.futures، اگر در یکی از نخ‌ها خطایی رخ دهد، برنامه کرش نمی‌کند. در عوض، آن خطا ذخیره شده و زمانی که متد .result() را صدا می‌زنید، پرتاب (Raise) می‌شود.

مثال ۱: مدیریت خطا در map (استاتیک)

اگر از map استفاده کنید، خطاها در هنگام پیمایش iterator بروز می‌کنند.

# این کد ریسک دارد چون اگر یکی خطا دهد، کل حلقه for متوقف می‌شود
for result in executor.map(buggy_function, data):
    print(result)

مثال ۲: مدیریت امن خطا با submit (تعاملی)

بهترین روش حرفه‌ای، استفاده از بلاک try/except هنگام فراخوانی future.result() است.

نکات فنی و بهینه‌سازی

1. انتخاب max_workers:
   • برای ThreadPoolExecutor: معمولاً min(32, os.cpu_count() + 4) پیشنهاد می‌شود.
   • برای ProcessPoolExecutor: پیش‌فرض برابر با تعداد هسته‌های CPU است. بیشتر کردن آن باعث کندی ناشی از Context Switching می‌شود.
2. Deadlocks: اگر در داخل یک Future منتظر نتیجه Future دیگری باشید که در همان Executor است، ممکن است دچار بن‌بست (Deadlock) شوید.