Python-Thread-Pool/Prozess-Pool für gleichzeitige Programmierung

Einführung

Die Python-Standardbibliothek stellt uns Threading- und Multiprocessing-Module zur Verfügung, um entsprechenden Multithreading-/Multiprozesscode zu schreiben. Wenn das Projekt jedoch eine bestimmte Größe erreicht, kommt es häufig zur Erstellung/Zerstörung von Prozessen oder Threads sind sehr ressourcenintensiv. Ja, zu diesem Zeitpunkt müssen wir unseren eigenen Thread-Pool/Prozesspool schreiben, um Platz gegen Zeit zu tauschen. Aber ab Python 3.2 stellt uns die Standardbibliothek das Modul concurrent.futures zur Verfügung, das zwei Klassen bereitstellt: ThreadPoolExecutor und ProcessPoolExecutor, die eine weitere Abstraktion von Threading und Multiprocessing realisieren und das Schreiben von Thread-Pools/Prozessen erleichtern. Pool bietet direkte Unterstützung.

Executor und Future

concurrent.futures-Modul basiert auf Executor ist eine abstrakte Klasse und kann nicht direkt verwendet werden. Allerdings sind die beiden bereitgestellten Unterklassen ThreadPoolExecutor und ProcessPoolExecutor sehr nützlich. Wie der Name schon sagt, werden sie zum Erstellen von Thread-Pool- bzw. Prozess-Pool-Codes verwendet. Wir können die entsprechenden Aufgaben direkt in den Thread-Pool/Prozesspool stellen und müssen uns nicht um Deadlocks kümmern. Der Thread-Pool/Prozesspool plant dies automatisch für uns.

ZukunftIch glaube, dass Freunde mit Programmiererfahrung in Java und NodeJS mit diesem Konzept vertraut sein werden Sie können es als einen in der Zukunft abgeschlossenen Vorgang verstehen. Dies ist die Grundlage der asynchronen Programmierung. Wenn wir beispielsweise queue.get ausführen, kommt es zu einer Blockierung, bevor auf die Rückgabe des Ergebnisses gewartet wird, und die CPU kann nicht für andere Aufgaben freigegeben werden Future hilft uns, die Aufgabe während der Wartezeit abzuschließen. Bezüglich asynchroner E/A in Python können Sie nach dem Lesen dieses Artikels auf meine Coroutine/asynchrone E/A für die gleichzeitige Programmierung in Python verweisen.

P.S.: Wenn Sie immer noch bei Python2.x bleiben, installieren Sie bitte zuerst das Futures-Modul.

Pip-Install-Futures

Verwenden Sie Submit, um den Thread-Pool/Prozess-Pool zu betreiben

Lassen Sie uns zunächst das Konzept des Thread-Pools anhand des folgenden Codes verstehen

# example1.py
aus concurrent.futures import ThreadPoolExecutor
import time
def return_future_result(message):
time.sleep(2)
return message
pool = ThreadPoolExecutor(max_workers=2) # Erstellen Sie einen Pool, der bis zu 2 Aufgaben-Thread aufnehmen kann pool
future1 = pool.submit(return_future_result, ("hello")) # Eine Aufgabe zum Thread-Pool hinzufügen
future2 = pool.submit(return_future_result, ("world")) # Zum Thread-Pool A hinzufügen task
print(future1.done()) # Bestimmen Sie, ob Aufgabe1 endet
time.sleep(3)
print(future2.done()) # Bestimmen Sie, ob Aufgabe2 endet
print(future1. result ()) # Überprüfen Sie die von Task1 zurückgegebenen Ergebnisse
print(future2.result()) # Überprüfen Sie die von Task2 zurückgegebenen Ergebnisse
Lassen Sie es uns anhand der laufenden Ergebnisse analysieren. Wir verwenden die Submit-Methode, um eine Aufgabe zum Thread-Pool hinzuzufügen, und Submit gibt ein Future-Objekt zurück. Das Future-Objekt kann einfach als eine in der Zukunft abgeschlossene Operation verstanden werden. In der ersten Druckanweisung ist es offensichtlich, dass unsere Zukunft1 aufgrund von time.sleep(2) nicht abgeschlossen ist, da wir time.sleep(3) verwenden, um den Hauptthread anzuhalten. Wenn wir also die zweite Druckanweisung erreichen, wird unser Thread angezeigt pool Alle Aufgaben hier wurden erledigt.

ziwenxie :: ~ » python example1.py
False
True
hello
world
# Während der Ausführung des obigen Programms können wir den ps-Befehl durchschauen Drei Threads laufen gleichzeitig im Hintergrund
ziwenxie :: ~ » ps -eLf |.ziwenxie 8361 7557 8361 3 3 19:45 pts/0 00:00:00 python example1.py
ziwenxie 8361 7557 8362 0 3 19:45 pts/0 00:00:00 python example1.py
ziwenxie 8361 7557 8363 0 3 19:45 pts/0 00:00:00 python example1.py
oben Der Code unseres Es kann auch in Form eines Prozesspools umgeschrieben werden. Die API und der Thread-Pool sind genau gleich, daher werde ich nicht auf Details eingehen.

# example2.py
from concurrent.futures import ProcessPoolExecutor
import time
def return_future_result(message):
time.sleep(2)
return message
pool = ProcessPoolExecutor(max_workers=2)
future1 = pool.submit(return_future_result, ("hello"))
future2 = pool.submit(return_future_result, ("world"))
print(future1.done ())
time.sleep(3)
print(future2.done())
print(future1.result())
print(future2.result())
Das Folgende ist Running results

ziwenxie :: ~ » python example2.py
False
True
hello
world
ziwenxie :: ~ » ps -eLf | > ziwenxie 8560 7557 8560 3 3 19:53 pts/0 00:00:00 python example2.py
ziwenxie 8560 7557 8563 0 3 19:53 pts/0 00:00:00 python example2.py
ziwenxie 8560 7557 8564 0 3 19:53 Punkte/0 00:00:00 Python-Beispiel2.py
ziwenxie 8561 8560 8561 0 1 19:53 Punkte/0 00:00:00 Python-Beispiel2.py
zi wenxie 8562 8560 8562 0 1 19:53 pts/0 00:00:00 python example2.py


Verwenden Sie Map/Warten, um den Thread-Pool/Prozess-Pool zu betreiben

Zusätzlich zum Senden, Execuor stellt uns auch eine Kartenmethode zur Verfügung, die der integrierten Kartenverwendung ähnelt. Vergleichen wir den Unterschied zwischen den beiden anhand von zwei Beispielen.

使用submit操作回顾

# example3.py
import concurrent.futures
import urllib.request
URLS = ['http://httpbin.org', 'http: //example.com/', 'https://api.github.com/']
def load_url(url, timeout):
   mit urllib.request.urlopen(url, timeout=timeout) als Verbindung :
       return conn.read()
# Wir können eine with-Anweisung verwenden, um sicherzustellen, dass Threads umgehend bereinigt werden
mit concurrent.futures.ThreadPoolExecutor(max_workers=3) als Executor:
   # Starten Sie die Laden Sie Vorgänge und markieren Sie jede Zukunft mit ihrer URL
   future_to_url = {executor.submit(load_url, url, 60): URL für URL in URLS}
   für Zukunft in concurrent.futures.as_completed(future_to_url):
url = future_to_url[future]
       try:
           data = future.result()
       außer Ausnahme als exc:
           print('%r hat eine Ausnahme generiert: %s' % (url, exc) )
       else:
           print('%r page is %d bytes' % (url, len(data)))
从运行结果可以看出，as_completed不是按照照照URLS列表元素的顺序返回的。

ziwenxie :: ~ » python example3.py
'http://example.com/' Seite ist 1270 Byte
'https://api.github .com/'-Seite ist 2039 Bytes
'http://httpbin.org'-Seite ist 12150 Bytes

使用map

# example4.py
gleichzeitig importieren .futures
import urllib.request
URLS = ['http://httpbin.org', 'http://example.com/', 'https://api.github.com/']
def load_url(url):
   with urllib.request.urlopen(url, timeout=60) as conn:
       return conn.read()
# Wir können eine with-Anweisung verwenden, um sicherzustellen, dass Threads vorhanden sind Prompt bereinigt
mit concurrent.futures.ThreadPoolExecutor(max_workers=3) als Executor:
   für URL, Daten in zip(URLS, executor.map(load_url, URLS)):
       print('%r Seite ist %d Bytes' % (URL, Länge(Daten)))的代码更加简洁直观，我们可以根据具体的需求任选一种。
ziwenxie :: ~ » python example4.py'http://httpbin.org' Seite ist 12150 Bytes
'http: //example.com/' Seite ist 1270 Bytes
'https://api.github.com/' Seite ist 2039 Bytes


第三种选择wait

wait方法接会返回一个tuple(元组), tuple中包含两个set(集合),一个是completed(已完成的)另外一个是uncompleted(未完成的)。使用wait方法的一个优势就是获得更大的自由度, 它接收三个参数FIRST_COMPLETED, FIRST_EXCEPTION. 和ALL_COMPLETE, 默认设置为ALL_COMPLETED子来看一下三个参数的区别

von concurrent.futures import ThreadPoolExecutor, wait, as_completed

from time import sleep
from random import randint
def return_after_random_secs(num):
   sleep(randint(1, 5))
   return „Return of {}“ .format(num)
pool = ThreadPoolExecutor(5)
futures = []
for x in range(5):
   futures.append(pool.submit(return_after_random_secs, x))
print(wait(futures))
# print(wait(futures, timeout=None, return_when='FIRST_COMPLETED'))
Alles abgeschlossen所有任务都完成。

ziwenxie :: ~ » python example5.py
DoneAndNotDoneFutures(done={
,
,
,
,
}, not_done = set ()) example5.py
DoneAndNotDoneFutures(done= {
,
,
},
not_done={,
})


思考题

写一个小程序对比Multiprocessing .pool(ThreadPool) und „ProcessPollExecutor(ThreadPoolExecutor)“ sind eine neue Version von Future.

Das obige ist der detaillierte Inhalt vonPython-Thread-Pool/Prozess-Pool für gleichzeitige Programmierung. Für weitere Informationen folgen Sie bitte anderen verwandten Artikeln auf der PHP chinesischen Website!