Współbieżność w Pythonie - pula procesów
Pula procesów może być tworzona i używana w ten sam sposób, w jaki stworzyliśmy i wykorzystaliśmy pulę wątków. Pula procesów może być zdefiniowana jako grupa procesów wstępnie zainicjowanych i bezczynnych, które są gotowe do wykonania pracy. Tworzenie puli procesów jest preferowane w stosunku do tworzenia instancji nowych procesów dla każdego zadania, gdy potrzebujemy wykonać dużą liczbę zadań.
Moduł Pythona - Concurrent.futures
Biblioteka standardowa Pythona zawiera moduł o nazwie concurrent.futures. Ten moduł został dodany w Pythonie 3.2, aby zapewnić programistom interfejs wysokiego poziomu do uruchamiania zadań asynchronicznych. Jest to warstwa abstrakcji znajdująca się na wierzchu modułów obsługi wątków i wieloprocesorowych Pythona, zapewniająca interfejs do uruchamiania zadań przy użyciu puli wątków lub procesów.
W kolejnych sekcjach przyjrzymy się różnym podklasom modułu concurrent.futures.
Klasa egzekutora
Executor jest abstrakcyjną klasą klasy concurrent.futuresModuł Pythona. Nie można go używać bezpośrednio i musimy użyć jednej z następujących konkretnych podklas -
- ThreadPoolExecutor
- ProcessPoolExecutor
ProcessPoolExecutor - konkretna podklasa
Jest to jedna z konkretnych podklas klasy Executor. Korzysta z multi-przetwarzania i otrzymujemy pulę procesów do składania zadań. Ta pula przypisuje zadania do dostępnych procesów i planuje ich uruchomienie.
Jak stworzyć ProcessPoolExecutor?
Z pomocą concurrent.futures moduł i jego konkretną podklasę Executor, możemy łatwo stworzyć pulę procesów. W tym celu musimy skonstruować plikProcessPoolExecutorz liczbą procesów, które chcemy w puli. Domyślnie jest to liczba 5. Następnie następuje przesłanie zadania do puli procesów.
Przykład
Rozważymy teraz ten sam przykład, którego użyliśmy podczas tworzenia puli wątków, jedyną różnicą jest to, że teraz będziemy używać ProcessPoolExecutor zamiast ThreadPoolExecutor .
from concurrent.futures import ProcessPoolExecutor
from time import sleep
def task(message):
sleep(2)
return message
def main():
executor = ProcessPoolExecutor(5)
future = executor.submit(task, ("Completed"))
print(future.done())
sleep(2)
print(future.done())
print(future.result())
if __name__ == '__main__':
main()Wynik
False
False
CompletedW powyższym przykładzie ProcessPoolExecutorzostał zbudowany z 5 wątków. Następnie zadanie, które będzie czekało 2 sekundy przed przekazaniem komunikatu, jest przesyłane do wykonawcy puli procesów. Jak widać z danych wyjściowych, zadanie nie kończy się przed upływem 2 sekund, więc pierwsze wywołaniedone()zwróci wartość False. Po 2 sekundach zadanie jest wykonane, a wynik na przyszłość uzyskujemy dzwoniąc doresult() metoda na nim.
Tworzenie wystąpienia ProcessPoolExecutor - Menedżer kontekstu
Innym sposobem na utworzenie instancji ProcessPoolExecutor jest użycie menedżera kontekstu. Działa podobnie do metody zastosowanej w powyższym przykładzie. Główną zaletą używania menedżera kontekstu jest to, że wygląda dobrze składniowo. Tworzenie instancji można wykonać za pomocą następującego kodu -
with ProcessPoolExecutor(max_workers = 5) as executorPrzykład
Aby lepiej zrozumieć, bierzemy ten sam przykład, który był używany podczas tworzenia puli wątków. W tym przykładzie musimy zacząć od zaimportowania plikuconcurrent.futuresmoduł. Następnie funkcja o nazwieload_url()zostanie utworzony, który załaduje żądany adres URL. PlikProcessPoolExecutorjest następnie tworzony z 5 liczbą wątków w puli. ProcesPoolExecutorzostał wykorzystany jako menedżer kontekstu. Możemy uzyskać wynik w przyszłości, dzwoniąc doresult() metoda na nim.
import concurrent.futures
from concurrent.futures import ProcessPoolExecutor
import urllib.request
URLS = ['http://www.foxnews.com/',
'http://www.cnn.com/',
'http://europe.wsj.com/',
'http://www.bbc.co.uk/',
'http://some-made-up-domain.com/']
def load_url(url, timeout):
with urllib.request.urlopen(url, timeout = timeout) as conn:
return conn.read()
def main():
with concurrent.futures.ProcessPoolExecutor(max_workers=5) as executor:
future_to_url = {executor.submit(load_url, url, 60): url for url in URLS}
for future in concurrent.futures.as_completed(future_to_url):
url = future_to_url[future]
try:
data = future.result()
except Exception as exc:
print('%r generated an exception: %s' % (url, exc))
else:
print('%r page is %d bytes' % (url, len(data)))
if __name__ == '__main__':
main()Wynik
Powyższy skrypt w języku Python wygeneruje następujące dane wyjściowe -
'http://some-made-up-domain.com/' generated an exception: <urlopen error [Errno 11004] getaddrinfo failed>
'http://www.foxnews.com/' page is 229476 bytes
'http://www.cnn.com/' page is 165323 bytes
'http://www.bbc.co.uk/' page is 284981 bytes
'http://europe.wsj.com/' page is 967575 bytesUżycie funkcji Executor.map ()
Python map()funkcja jest szeroko stosowana do wykonywania wielu zadań. Jednym z takich zadań jest zastosowanie określonej funkcji do każdego elementu w elementach iterable. Podobnie możemy zmapować wszystkie elementy iteratora na funkcję i przesłać je jako niezależne zadania doProcessPoolExecutor. Aby to zrozumieć, rozważ następujący przykład skryptu w języku Python.
Przykład
Rozważymy ten sam przykład, którego użyliśmy podczas tworzenia puli wątków przy użyciu Executor.map()funkcjonować. W przykładzie podanym poniżej funkcja map służy do zastosowaniasquare() funkcji do każdej wartości w tablicy wartości.
from concurrent.futures import ProcessPoolExecutor
from concurrent.futures import as_completed
values = [2,3,4,5]
def square(n):
return n * n
def main():
with ProcessPoolExecutor(max_workers = 3) as executor:
results = executor.map(square, values)
for result in results:
print(result)
if __name__ == '__main__':
main()Wynik
Powyższy skrypt w języku Python wygeneruje następujące dane wyjściowe
4
9
16
25Kiedy używać ProcessPoolExecutor i ThreadPoolExecutor?
Teraz, gdy przestudiowaliśmy obie klasy Executorów - ThreadPoolExecutor i ProcessPoolExecutor, musimy wiedzieć, kiedy użyć którego executora. Musimy wybrać ProcessPoolExecutor w przypadku obciążeń związanych z procesorem i ThreadPoolExecutor w przypadku obciążeń związanych z we / wy.
Jeśli używamy ProcessPoolExecutor, wtedy nie musimy się martwić o GIL, ponieważ wykorzystuje on wieloprocesorowość. Co więcej, czas wykonania będzie krótszy w porównaniu zThreadPoolExecution. Aby to zrozumieć, rozważ następujący przykład skryptu w języku Python.
Przykład
import time
import concurrent.futures
value = [8000000, 7000000]
def counting(n):
start = time.time()
while n > 0:
n -= 1
return time.time() - start
def main():
start = time.time()
with concurrent.futures.ProcessPoolExecutor() as executor:
for number, time_taken in zip(value, executor.map(counting, value)):
print('Start: {} Time taken: {}'.format(number, time_taken))
print('Total time taken: {}'.format(time.time() - start))
if __name__ == '__main__':
main()Wynik
Start: 8000000 Time taken: 1.5509998798370361
Start: 7000000 Time taken: 1.3259999752044678
Total time taken: 2.0840001106262207
Example- Python script with ThreadPoolExecutor:
import time
import concurrent.futures
value = [8000000, 7000000]
def counting(n):
start = time.time()
while n > 0:
n -= 1
return time.time() - start
def main():
start = time.time()
with concurrent.futures.ThreadPoolExecutor() as executor:
for number, time_taken in zip(value, executor.map(counting, value)):
print('Start: {} Time taken: {}'.format(number, time_taken))
print('Total time taken: {}'.format(time.time() - start))
if __name__ == '__main__':
main()Wynik
Start: 8000000 Time taken: 3.8420000076293945
Start: 7000000 Time taken: 3.6010000705718994
Total time taken: 3.8480000495910645Na podstawie wyników obu powyższych programów możemy zobaczyć różnicę w czasie wykonywania podczas używania ProcessPoolExecutor i ThreadPoolExecutor.