آموزش ماژول concurrent.futures – کلاس ThreadPoolExecutor

کلاس ThreadPoolExecutor

کلاس ThreadPoolExecutor از ماژول concurrent.futures برای مدیریت یک «استخر» از تردها به کار می‌رود. به‌جای اینکه خودتان دستی برای هر کار یک Thread بسازید و مدیریت کنید، فقط کارها را به executor می‌سپارید و آن تعداد مشخصی ترد را برایتان اجرا و بازیافت می‌کند. جالب اینجاست که API این کلاس دقیقاً مثل ProcessPoolExecutor است؛ پس اگر یکی را بلد باشید، دیگری را هم بلدید.

اجرای کارها با map

با map یک تابع را روی همه‌ی عناصر یک فهرست اجرا می‌کنیم. پارامتر max_workers تعداد تردهای هم‌زمان را محدود می‌کند:

import time
from concurrent.futures import ThreadPoolExecutor

def task(name):
    print(f"Task {name} started")
    time.sleep(1)
    print(f"Task {name} finished")

with ThreadPoolExecutor(max_workers=2) as executor:
    executor.map(task, [1, 2, 3, 4, 5])

import time
from concurrent.futures import ThreadPoolExecutor

def task(name):
    print(f"Task {name} started")
    time.sleep(1)
    print(f"Task {name} finished")

with ThreadPoolExecutor(max_workers=2) as executor:
    executor.map(task, [1, 2, 3, 4, 5])

چون max_workers=2 است، در هر لحظه فقط دو کار اجرا می‌شوند و تا وقتی یکی تمام نشود، کار بعدی شروع نمی‌شود. خروجی چیزی شبیه این است:

Task 1 started
Task 2 started
Task 1 finished
Task 3 started
Task 2 finished
Task 4 started
Task 3 finished
Task 5 started
Task 4 finished
Task 5 finished

گرفتن نتایج با submit و Future

اگر تابع شما مقدار return دارد و می‌خواهید خروجی‌ها را جمع کنید، از submit استفاده کنید که یک Future برمی‌گرداند؛ سپس با .result() نتیجه‌ی هرکدام را می‌گیرید:

import time
from concurrent.futures import ThreadPoolExecutor

def task(name):
    print(f"Task {name} started")
    time.sleep(1)
    print(f"Task {name} finished")
    return f"Result of task {name}"

with ThreadPoolExecutor(max_workers=2) as executor:
    # submit tasks and get futures
    futures = [executor.submit(task, i) for i in [1, 2, 3, 4]]

    # retrieve results from the futures
    results = [future.result() for future in futures]

    print("Results:", results)

import time
from concurrent.futures import ThreadPoolExecutor

def task(name):
    print(f"Task {name} started")
    time.sleep(1)
    print(f"Task {name} finished")
    return f"Result of task {name}"

with ThreadPoolExecutor(max_workers=2) as executor:
    # submit tasks and get futures
    futures = [executor.submit(task, i) for i in [1, 2, 3, 4]]

    # retrieve results from the futures
    results = [future.result() for future in futures]

    print("Results:", results)

دریافت نتایج به ترتیب اتمام با as_completed

اگر می‌خواهید نتیجه‌ی هر کار را به‌محض آماده‌شدن پردازش کنید (نه به ترتیب ورودی)، از as_completed استفاده کنید:

from concurrent.futures import ThreadPoolExecutor, as_completed

def task(name):
    return f"Result of task {name}"

with ThreadPoolExecutor(max_workers=3) as executor:
    futures = [executor.submit(task, i) for i in range(5)]
    for future in as_completed(futures):
        print(future.result())

from concurrent.futures import ThreadPoolExecutor, as_completed

def task(name):
    return f"Result of task {name}"

with ThreadPoolExecutor(max_workers=3) as executor:
    futures = [executor.submit(task, i) for i in range(5)]
    for future in as_completed(futures):
        print(future.result())

مثال واقعی: دانلود هم‌زمان چند صفحه

مهم‌ترین کاربرد ThreadPoolExecutor کارهای I/O-bound است؛ مثل دانلود چند صفحه، فراخوانی چند API یا خواندن چند فایل. چون تردها بیشترِ وقت منتظر شبکه‌اند، هشت دانلود دو‌ثانیه‌ای به‌جای ۱۶ ثانیه، حدود ۲ ثانیه طول می‌کشد:

import time
from concurrent.futures import ThreadPoolExecutor

def fetch(url):
    time.sleep(2)                       # simulate a network request
    return f"{url} done"

urls = [f"site{i}.com" for i in range(8)]

start = time.time()
with ThreadPoolExecutor(max_workers=8) as executor:
    results = list(executor.map(fetch, urls))
print("took %.1fs" % (time.time() - start))   # ~2s, not ~16s

import time
from concurrent.futures import ThreadPoolExecutor

def fetch(url):
    time.sleep(2)                       # simulate a network request
    return f"{url} done"

urls = [f"site{i}.com" for i in range(8)]

start = time.time()
with ThreadPoolExecutor(max_workers=8) as executor:
    results = list(executor.map(fetch, urls))
print("took %.1fs" % (time.time() - start))   # ~2s, not ~16s

در دنیای واقعی به‌جای time.sleep، یک درخواست شبکه‌ی واقعی (مثلاً با کتابخانه‌ی requests) قرار می‌گیرد.

ThreadPoolExecutor در برابر ProcessPoolExecutor

هر دو API یکسانی دارند (map، submit، as_completed) و جابه‌جایی بینشان فقط تغییر نام کلاس است. اما انتخاب درست به نوع کار بستگی دارد: برای کارهای I/O-bound (شبکه، فایل، دیتابیس) از ThreadPoolExecutor استفاده کنید؛ برای کارهای CPU-bound (محاسبات سنگین) از ProcessPoolExecutor، چون تردها به‌خاطر GIL در کارهای CPU سرعتی به‌دست نمی‌آورند.

جمع‌بندی

• ThreadPoolExecutor یک استخر از تردها می‌سازد و کارها را با تعداد محدود (max_workers) به‌صورت هم‌زمان اجرا می‌کند.
• map برای اجرای یک تابع روی یک فهرست، submit برای گرفتن Future و جمع‌آوری نتایج با .result().
• با as_completed نتایج را به ترتیب اتمام دریافت کنید.
• API آن با ProcessPoolExecutor یکسان است؛ برای I/O از تردها و برای CPU از پردازه‌ها استفاده کنید.