آموزش کار با ThreadPoolExecutor در پایتون ۳ (پردازش موازی و افزایش سرعت)

کیانا۱۴۰۵-۰۵-۰۲

0 28 خواندن این مطلب 4 دقیقه زمان میبرد

مقدمه

نخ‌ها (Threads) در پایتون یکی از روش‌های اصلی پردازش موازی (Parallelism) هستند که به برنامه شما اجازه می‌دهند چند عملیات را به‌صورت هم‌زمان اجرا کند. اگرچه پردازش موازی در پایتون با فرآیندهای چندگانه (Multiprocessing) نیز امکان‌پذیر است، اما Threadها به‌طور ویژه برای افزایش سرعت برنامه‌هایی که درگیر عملیات ورودی/خروجی (I/O-Bound) هستند، عالی عمل می‌کنند.

از نمونه‌های بارز عملیات I/O-Bound می‌توان به ارسال درخواست‌های وب (Web Requests) و خواندن داده‌ها از فایل اشاره کرد. برعکس، عملیات سنگین پردازشی (CPU-Bound) مثل محاسبات ریاضی پیچیده، سود چندانی از Threadها نمی‌برند (این موضوع به دلیل وجود قفل جهانی مفسر یا GIL در پایتون است).

پایتون ۳ ابزار قدرتمندی به نام ThreadPoolExecutor در ماژول concurrent.futures ارائه می‌دهد که مدیریت و اجرای کدهای چند‌نخی را بسیار ساده می‌سازد.

در این مقاله از وبلاگ پارمین کلود، با استفاده از ThreadPoolExecutor سرعت ارسال درخواست‌های شبکه را به شکلی چشم‌گیر افزایش می‌دهیم.

پیش‌نیازها

یک محیط برنامه‌نویسی پایتون ۳ (مثلاً روی سرور ابری Ubuntu 24.04 در پارمین کلود).
نصب کتابخانه requests.

💡 نکته نصب پکیج با سرعت بالا: برای نصب کتابخانه requests و جهت دور زدن تحریم‌ها و افزایش سرعت دانلود بر روی سرورهای ابری، می‌توانید از میرور اختصاصی PyPI پارمین کلود استفاده کنید:
Bash
pip install --index-url https://pypi.parmincloud.ir/simple/ requests

گام اول — تعریف تابعی برای اجرا در Threadها

ابتدا تابعی می‌نویسیم که وجود یا عدم وجود یک صفحه در ویکی‌پدیا را بررسی کند. این کار شامل ارسال یک درخواست شبکه‌ای (عملیات I/O-Bound) است.

یک فایل به نام wiki_page_function.py بسازید:

Bash

nano wiki_page_function.py

کدهای زیر را درون آن قرار دهید:

Python

import requests

def get_wiki_page_existence(wiki_page_url, timeout=10):
    response = requests.get(url=wiki_page_url, timeout=timeout)

    page_status = "unknown"
    if response.status_code == 200:
        page_status = "exists"
    elif response.status_code == 404:
        page_status = "does not exist"

    return wiki_page_url + " - " + page_status

# تست تابع با یک آدرس مشخص
url = "https://en.wikipedia.org/wiki/Ocean"
print(get_wiki_page_existence(wiki_page_url=url))

برنامه را اجرا کنید:

Bash

python3 wiki_page_function.py

خروجی زیر نشان می‌دهد که صفحه وجود دارد:

Plaintext

https://en.wikipedia.org/wiki/Ocean - exists

هشدار مهم درباره هم‌روندی: به‌طور کلی، به‌اشتراک‌گذاری اشیاء یا متغیرها بین Threadهای مختلف در پایتون بدون کنترل هم‌زمانی خطرساز است. تابعی مانند get_wiki_page_existence که یک کار ایزوله انجام داده و با Threadهای دیگر متغیر مشترکی ندارد، بهترین گزینه برای اجرای چند‌نخی است.

گام دوم — اجرای هم‌زمان تابع با ThreadPoolExecutor

حالا که تابع خود را آماده کرده‌ایم، از ThreadPoolExecutor استفاده می‌کنیم تا چندین آدرس وب را به‌صورت هم‌زمان و موازی بررسی کنیم.

فایل wiki_page_function.py را ویرایش کرده و کدهای زیر را جایگزین کنید:

Python

import requests
import concurrent.futures

def get_wiki_page_existence(wiki_page_url, timeout=10):
    response = requests.get(url=wiki_page_url, timeout=timeout)

    page_status = "unknown"
    if response.status_code == 200:
        page_status = "exists"
    elif response.status_code == 404:
        page_status = "does not exist"

    return wiki_page_url + " - " + page_status

# لیستی از آدرس‌های مورد نظر برای بررسی
wiki_page_urls = [
    "https://en.wikipedia.org/wiki/Ocean",
    "https://en.wikipedia.org/wiki/Island",
    "https://en.wikipedia.org/wiki/this_page_does_not_exist",
    "https://en.wikipedia.org/wiki/Shark",
]

# ساخت Executor برای مدیریت نخ‌ها
with concurrent.futures.ThreadPoolExecutor() as executor:
    futures = []
    for url in wiki_page_urls:
        futures.append(executor.submit(get_wiki_page_existence, wiki_page_url=url))
    
    # دریافت نتایج به محض آماده شدن هر نخ
    for future in concurrent.futures.as_completed(futures):
        print(future.result())

کد چگونه کار می‌کند؟

ماژول concurrent.futures ابزار ThreadPoolExecutor را وارد برنامه می‌کند.
عبارت with یک نمونه از Executor می‌سازد که پس از اتمام کار، Threadها را به‌صورت خودکار بسته‌بندی و آزاد می‌کند.
متد executor.submit() کار بررسی هر URL را به یک Thread مستقل می‌سپارد و یک شیء Future برمی‌گرداند.
تابع as_completed() نتایج را به محض آماده شدن (بدون توجه به ترتیب ارسال) برمی‌گرداند.

با اجرای مجدد کد، خروجی مشابه زیر را مشاهده می‌کنید:

Plaintext

https://en.wikipedia.org/wiki/Island - exists
https://en.wikipedia.org/wiki/Ocean - exists
https://en.wikipedia.org/wiki/this_page_does_not_exist - does not exist
https://en.wikipedia.org/wiki/Shark - exists

گام سوم — مدیریت خطاهایی که در Thread رخ می‌دهند

اگر در زمان اجرای یک Thread خطایی رخ دهد (مثلاً Time-out شبکه)، ThreadPoolExecutor این استثنا (Exception) را ذخیره می‌کند و زمانی که متد future.result() را فراخوانی می‌کنید، آن را رخ می‌دهد.

برای تست این موضوع، تایم‌آوت درخواست‌ها را به 0.00001 ثانیه کاهش می‌دهیم تا خطا رخ دهد:

Python

import requests
import concurrent.futures

def get_wiki_page_existence(wiki_page_url, timeout=10):
    response = requests.get(url=wiki_page_url, timeout=timeout)

    page_status = "unknown"
    if response.status_code == 200:
        page_status = "exists"
    elif response.status_code == 404:
        page_status = "does not exist"

    return wiki_page_url + " - " + page_status

wiki_page_urls = [
    "https://en.wikipedia.org/wiki/Ocean",
    "https://en.wikipedia.org/wiki/Island",
    "https://en.wikipedia.org/wiki/this_page_does_not_exist",
    "https://en.wikipedia.org/wiki/Shark",
]

with concurrent.futures.ThreadPoolExecutor() as executor:
    futures = []
    for url in wiki_page_urls:
        futures.append(
            executor.submit(
                get_wiki_page_existence, wiki_page_url=url, timeout=0.00001
            )
        )
    for future in concurrent.futures.as_completed(futures):
        try:
            print(future.result())
        except requests.ConnectTimeout:
            print("خطای تایم‌آوت: زمان اتصال به پایان رسید.")

خروجی:

Plaintext

خطای تایم‌آوت: زمان اتصال به پایان رسید.
خطای تایم‌آوت: زمان اتصال به پایان رسید.
خطای تایم‌آوت: زمان اتصال به پایان رسید.
خطای تایم‌آوت: زمان اتصال به پایان رسید.

به این ترتیب، با قرار دادن future.result() درون بلوک try/except می‌توانید خطاهای رخ داده درون Threadها را به‌راحتی مدیریت کنید.

گام چهارم — مقایسه سرعت اجرای برنامه با و بدون Thread

برای درک میزان تاثیر ThreadPoolExecutor در سرعت اجرا، ۵۰ آدرس مختلف از ویکی‌پدیا را یک‌بار به صورت تک‌نخی (معمولی) و یک‌بار با استفاده از Threadها اجرا و زمان را اندازه‌گیری می‌کنیم.

الف) اجرای بدون Thread (تک‌نخی):

Python

import time
import requests

def get_wiki_page_existence(wiki_page_url, timeout=10):
    response = requests.get(url=wiki_page_url, timeout=timeout)
    page_status = "exists" if response.status_code == 200 else "does not exist"
    return wiki_page_url + " - " + page_status

wiki_page_urls = ["https://en.wikipedia.org/wiki/" + str(i) for i in range(50)]

print("در حال اجرای برنامه بدون Thread:")
start_time = time.time()

for url in wiki_page_urls:
    print(get_wiki_page_existence(wiki_page_url=url))

print("زمان اجرا بدون Thread:", time.time() - start_time)

خروجی نمونه:

Plaintext

Without threads time: 5.803015232086182

(بررسی ۵۰ لینک حدود ۵٫۸ ثانیه طول کشید).

ب) اجرای با ThreadPoolExecutor (چند‌نخی):

Python

import time
import requests
import concurrent.futures

def get_wiki_page_existence(wiki_page_url, timeout=10):
    response = requests.get(url=wiki_page_url, timeout=timeout)
    page_status = "exists" if response.status_code == 200 else "does not exist"
    return wiki_page_url + " - " + page_status

wiki_page_urls = ["https://en.wikipedia.org/wiki/" + str(i) for i in range(50)]

print("در حال اجرای برنامه با ThreadPoolExecutor:")
start_time = time.time()

with concurrent.futures.ThreadPoolExecutor() as executor:
    futures = [executor.submit(get_wiki_page_existence, wiki_page_url=url) for url in wiki_page_urls]
    for future in concurrent.futures.as_completed(futures):
        print(future.result())

print("زمان اجرا با Thread:", time.time() - start_time)

خروجی نمونه:

Plaintext

Threaded time: 1.2201685905456543

همان‌طور که می‌بینید، زمان اجرا از ۵٫۸ ثانیه به ۱٫۲ ثانیه کاهش یافت! این یعنی برنامه نزدیک به ۵ برابر سریع‌تر اجرا شده است.

جمع‌بندی

استفاده از ThreadPoolExecutor در پایتون ۳ یکی از ساده‌ترین و مؤثرترین روش‌ها برای بهینه‌سازی کدهایی است که با شبکه، پایگاه‌داده یا فایل‌ها (عملیات I/O) سرکار دارند. شما با استفاده از این ابزار می‌توانید بدون درگیر شدن با پیچیدگی‌های دستی ساخت نخ‌ها، کارایی کدهای خود را روی سرورهای ابری به‌طور شگفت‌انگیزی ارتقا دهید.

از همراهی شما با پارمین کلود سپاسگزاریم.

کیانا۱۴۰۵-۰۵-۰۲

0 28 خواندن این مطلب 4 دقیقه زمان میبرد