concurrent-futures - بستیکال

برنامه نویسی موازی در پایتون با concurrent.futures

امروز میخوایم با ماژول concurrent.futures برنامه نویسی موازی انجام بدیم، این ماژول در نسخه ۳ پایتون موجود هست.

قبل از خوندن ادامه متن، حتما نوشته فکر میکنید پایتون میتونه دو تا کار رو همزمان انجام بده؟ رو بخونید.

برنامه ای رو در نظر بگیرید که ۱۰۰ صفحه وب رو میخواد پردازش کنه یعنی ۱۰۰ تا thread باید بسازید یا ۱۰۰ تا process حالا اینکه اینها به صورت همزمان اجرا بشن بستگی به سخت افزارتون داره (در این مورد خاص که خیلی ساده هست حتما میشه)

استفاده از ترد

برای اجرای این برنامه با ترد به صورت زیر عمل میکنیم

from concurrent.futures import ThreadPoolExecutor
import requests 

urls = ['https://github.com?v=' + str(i) for i in range(100)]
with ThreadPoolExecutor(max_workers=10) as e:
    for url in urls:
        e.submit(requests.get, url)
    # tasks not completed yet
# now all tasks are completed

from concurrent.futures import ThreadPoolExecutor

import requests

urls = ['https://github.com?v=' + str(i) for i in range(100)]

with ThreadPoolExecutor(max_workers=10) as e:

for url in urls:

e.submit(requests.get, url)

# tasks not completed yet

# now all tasks are completed

وقتی از کانتکس منیجر with استفاده میکنید به شما اطمینان داده میشه که وقتی برنامه از with خارج میشه که، کار تمام تسک ها تموم شده باشه.

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

from concurrent.futures import ThreadPoolExecutor, as_completed
import requests

urls = ['https://github.com?v=' + str(i) for i in range(100)]
with ThreadPoolExecutor(max_workers=10) as e:
    futures = [e.submit(requests.get, url) for url in urls]
    for future in as_completed(futures):
        print(future.result())

from concurrent.futures import ThreadPoolExecutor, as_completed

import requests

urls = ['https://github.com?v=' + str(i) for i in range(100)]

with ThreadPoolExecutor(max_workers=10) as e:

futures = [e.submit(requests.get, url) for url in urls]

for future in as_completed(futures):

print(future.result())

ThreadPoolExecutor یک مقدار max_workers میگیره که مشخص میکنه با چه تعداد ترد این عملیات ها رو به صورت همزمان اجرا کنه اگه مقداری ندید مقدار پیش فرضش تعداد Processor های CPU ضرب در ۵ خواهد بود.

با کد زیر هم میتونید تعداد Processor های CPU رو به دست بیارید

from multiprocessing import cpu_count
no_of_processors = cpu_count()

1 2	from multiprocessing import cpu_count no_of_processors = cpu_count()

اینجا هم یک روش دیگه برای به دست آوردن تعداد Processor های CPU معرفی شده.

در مثال بالا با استفاده از submit پردازش ۱۰۰ صفحه رو شروع میکنیم و در مرحله بعد برای اینکه خروجی رو ببینیم از تابع as_completed استفاده میکنیم این تابع آرایه ای از متغیر future رو میگیره و تا تموم شدن تمام فرایندها اجرای کد رو متوقف میکنه.

متغیر future یک متغیر خاص هست و همونطور که از اسمش معلوم هنوز مقدار کامل نداره

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

with ThreadPoolExecutor(max_workers=10) as e:
    results = e.map(requests.get, urls)
    # tasks are completed here
    for result in results:
        print(result)

with ThreadPoolExecutor(max_workers=10) as e:

results = e.map(requests.get, urls)

# tasks are completed here

for result in results:

print(result)

وقتی از تابع map استفاده میکنید تا تموم شدن تمام فرایندها برنامه به خط بعدی نمیره یعنی قبل از رسیدن برنامه به حلقه تمام فرآیندها انجام شده

تفاوت مهم بین submit و map این هست که وقتی از submit و در ادامه اون از as_completed استفاده میکنید نتایجی رو زودتر میگیرید که زودتر تموم شدن ولی در map نتایح رو بر اساس ترتیبی که دادید میگیرید برای مثال اگه ادرس ها رو در لیست (که در اونها ترتیب مقادیر حفظ میشه ) به map ارسال کرده باشید (مثل بالا) نتایج رو به همون ترتیب هم میگیرید.

البته استفاده از این دو، به نوع برنامه و کاری که میخواید انجام بدید بستگی داره

استفاده از پروسس

برای اینکه همین برنامه رو با پروسس اجرا کنیم به صورت زیر باید عمل کنیم

from concurrent.futures import ProcessPoolExecutor
import requests

urls = ['https://github.com?v=' + str(i) for i in range(100)]
with ProcessPoolExecutor(max_workers=10) as e:
    futures = [e.submit(requests.get, url) for url in urls]
    # tasks not completed yet 
    for future in as_completed(futures): 
        print(future.result())
# tasks are completed here

from concurrent.futures import ProcessPoolExecutor

import requests

urls = ['https://github.com?v=' + str(i) for i in range(100)]

with ProcessPoolExecutor(max_workers=10) as e:

futures = [e.submit(requests.get, url) for url in urls]

# tasks not completed yet

for future in as_completed(futures):

print(future.result())

# tasks are completed here

ProcessPoolExecutor یک مقدار max_workers میگیره که مشخص میکنه با چه تعداد پروسس این عملیات ها رو به صورت همزمان اجرا کنه اگه مقداری ندید مقدار پیش فرض تعداد Processor های CPU خواهد بود.

دقت کنید که منظور کل پروسس ها برای اجرای برنامه هست نه تعداد پروسس های همزمان.

روش دوم هم استفاده از تابع map هست

with ThreadPoolExecutor(max_workers=10) as e:
    results = e.map(requests.get, urls, 10)
    # tasks are completed here 
    for result in results: 
        print(result)

with ThreadPoolExecutor(max_workers=10) as e:

results = e.map(requests.get, urls, 10)

# tasks are completed here

for result in results:

print(result)

تابع map وقتی با پروسس اجرا بشه یک آرگومان دیگه به نام chunksize میگیره که مشخص میکنه در هر بار چند تا فرایند برای اجرا استارت بخوره.

مقدار پیش فرض این آرگومان ۱ هست یعنی دونه دونه اجرای تابع requests.get برای هر url پشت سر هم به صورت ترتیبی ارسال بشه اگه برای مثال ۱۰ بزارید تابع map به صورت همزمان ۱۰ تا ۱۰ تا آدرس میگیره و به صورت گروهی درخواست اجرای این تابع رو ارسال میکنه

تنظیم این مقدار وقتی از پروسس استفاده میکنید میتونه تاثیر زیادی در افزایش سرعت برنامتون داشته باشه. (دقت کنید که منظور ارسال درخواست اجرا هست نه خود اجرا)