Parallele Taskausführung mit concurrent.futures in Python

Übersicht über das Modul concurrent.futures

Das in Python 3.2 hinzugefügte Modul concurrent.futures bietet die Möglichkeit, mehrere Prozesse parallel auszuführen.

Es gibt andere Module in Python, die als Threading und Multiprocessing bezeichnet werden. Diese behandeln jedoch einen Thread-Prozess, während das Modul concurrent.futures mehrere Thread-Prozesse verarbeiten soll. ist.

Executor

Das Modul concurrent.futures verfügt über eine Executor -Klasse als abstrakte Klasse, und zwei Klassen werden als Implementierungsklassen bereitgestellt. Verwenden Sie eine der beiden Optionen, um parallele Aufgaben auszuführen.

• ThreadPoolExecutor --Verwenden Sie Threads, um parallele Aufgaben auszuführen.
• Geeignet für die parallele Ausführung von Prozessen, die die CPU nicht belasten, z. B. Netzwerkzugriff.
• ProcessPoolExecutor --Verwenden Sie Prozesse, um parallele Aufgaben auszuführen.
• Geeignet für parallele Ausführung wie CPU-intensive Berechnungsverarbeitung.

max_workers

Im Executor-Konstruktor wird die maximale Anzahl von Aufgaben, die gleichzeitig ausgeführt werden können, durch das Argument max_workers angegeben. Wenn Sie mehr Aufgaben zum Ausführen anfordern, als gleichzeitig ausgeführt werden können, werden die Aufgaben zur Warteschlange hinzugefügt, warten auf den Abschluss der anderen Aufgaben und werden dann nacheinander ausgeführt.

Methoden zum Ausführen von Aufgaben senden und zuordnen

Executor verfügt über die folgenden Methoden, um parallele Aufgaben auszuführen.

• submit
• Fügen Sie der Ausführungswarteschlange eine Aufgabe hinzu. --Wenn die laufende Aufgabe kleiner als max_workers ist, wird die hinzugefügte Aufgabe sofort ausgeführt.
• Das Rückgabewertobjekt Future bricht die Aufgabe ab und erfasst das Ausführungsergebnis.
• map
• Übergeben Sie die Ausführungsaufgabe mit dem Iterator.
• Der Rückgabewert ist ein Generator, um das Ausführungsergebnis der Aufgabe zu erhalten.

Stichprobe

Dies ist ein Beispiel mit concurrent.futures. Es ist in Python3.6 geschrieben, aber ich denke, dass es mit einer kleinen Änderung betrieben werden kann, wenn die Umgebung 3.2 oder höher ist.

• 01_thread.py
• Einfaches Beispiel mit ThreadPool Executor
• 02_future.py
• Beispiel für den Erhalt des Ergebnisses der Übermittlung
• 03_map.py
• Beispiel für das Hinzufügen von Aufgaben im Stapel / das Erfassen von Ergebnissen im Stapel mit Karte
• 04_process.py --Beispiel für die Ausführung einer umfangreichen Verarbeitung mit ProcessPoolExecutor

Ein einfaches Beispiel mit ThreadPoolExecutor

01_thread.py Führen Sie 5 Aufgaben in 2 Threads aus. Die Aufgabe besteht darin, nur 1 Sekunde zu schlafen.

python

def task(v):
    getLogger().info("%s start", v)
    time.sleep(1.0)
    getLogger().info("%s end", v)

def main():
    init_logger()
    getLogger().info("main start")
    with ThreadPoolExecutor(max_workers=2, thread_name_prefix="thread") as executor:
        for i in range(5):
            executor.submit(task, i)
        getLogger().info("submit end")
    getLogger().info("main end")

[2017-04-02 12:01:39,747] [MainThread] main start
[2017-04-02 12:01:39,748] [thread_0] 0 start
[2017-04-02 12:01:39,749] [thread_1] 1 start
[2017-04-02 12:01:39,750] [MainThread] submit end
[2017-04-02 12:01:40,755] [thread_0] 0 end
[2017-04-02 12:01:40,755] [thread_0] 2 start
[2017-04-02 12:01:40,756] [thread_1] 1 end
[2017-04-02 12:01:40,756] [thread_1] 3 start
[2017-04-02 12:01:41,761] [thread_0] 2 end
[2017-04-02 12:01:41,761] [thread_0] 4 start
[2017-04-02 12:01:41,761] [thread_1] 3 end
[2017-04-02 12:01:42,764] [thread_0] 4 end
[2017-04-02 12:01:42,765] [MainThread] main end

Erhalten Sie Ergebnisse mit Senden

02_future.py

Dies ist ein Beispiel für den Empfang des Ergebnisses von submit.

def task(v):
    getLogger().info("%s start", v)
    time.sleep(1.0)
    getLogger().info("%s end", v)
    return v * 2

def main():
    init_logger()
    getLogger().info("main start")
    with ThreadPoolExecutor(max_workers=2, thread_name_prefix="thread") as executor:
        futures = []
        for i in range(5):
            futures.append(executor.submit(task, i))
        getLogger().info("submit end")
        getLogger().info([f.result() for f in futures])
    getLogger().info("main end")

[2017-04-02 12:08:23,853] [MainThread] main start
[2017-04-02 12:08:23,854] [thread_0] 0 start
[2017-04-02 12:08:23,855] [thread_1] 1 start
[2017-04-02 12:08:23,856] [MainThread] submit end
[2017-04-02 12:08:24,856] [thread_0] 0 end
[2017-04-02 12:08:24,856] [thread_0] 2 start
[2017-04-02 12:08:24,857] [thread_1] 1 end
[2017-04-02 12:08:24,857] [thread_1] 3 start
[2017-04-02 12:08:25,863] [thread_0] 2 end
[2017-04-02 12:08:25,864] [thread_0] 4 start
[2017-04-02 12:08:25,864] [thread_1] 3 end
[2017-04-02 12:08:26,867] [thread_0] 4 end
[2017-04-02 12:08:26,868] [MainThread] [0, 2, 4, 6, 8]
[2017-04-02 12:08:26,868] [MainThread] main end

Beispiel für das Hinzufügen von Aufgaben im Stapel / das Erfassen von Ergebnissen im Stapel mit Karte

03_map.py Wenn Sie Aufgaben in einem Stapel verarbeiten möchten, ist das Schreiben mit map einfacher als mit submit.

def task(v):
    getLogger().info("%s start", v)
    time.sleep(1.0)
    getLogger().info("%s end", v)
    return v * 2

def main():
    init_logger()
    getLogger().info("main start")
    with ThreadPoolExecutor(max_workers=2, thread_name_prefix="thread") as executor:
        results = executor.map(task, range(5))
        getLogger().info("map end")
    getLogger().info(list(results))
    getLogger().info("main end")

[2017-04-02 12:10:03,997] [MainThread] main start
[2017-04-02 12:10:03,998] [thread_0] 0 start
[2017-04-02 12:10:04,000] [thread_1] 1 start
[2017-04-02 12:10:04,000] [MainThread] map end
[2017-04-02 12:10:05,005] [thread_0] 0 end
[2017-04-02 12:10:05,006] [thread_0] 2 start
[2017-04-02 12:10:05,006] [thread_1] 1 end
[2017-04-02 12:10:05,006] [thread_1] 3 start
[2017-04-02 12:10:06,007] [thread_0] 2 end
[2017-04-02 12:10:06,007] [thread_0] 4 start
[2017-04-02 12:10:06,007] [thread_1] 3 end
[2017-04-02 12:10:07,014] [thread_0] 4 end
[2017-04-02 12:10:07,014] [MainThread] [0, 2, 4, 6, 8]
[2017-04-02 12:10:07,014] [MainThread] main end

Beispiel für die Ausführung einer umfangreichen Verarbeitung mit ProcessPoolExecutor

04_process.py Zum Schluss ein Beispiel mit Process Pool Executor. Lassen Sie uns die Parameter ändern und den Unterschied in der Leistung sehen.

def task(params):
    (v, num_calc) = params
    a = float(v)
    for _ in range(num_calc):
        a = pow(a, a)
    return a

def main():
    init_logger()

    if len(sys.argv) != 5:
        print("usage: 05_process.py max_workers chunk_size num_tasks num_calc")
        sys.exit(1)
    (max_workers, chunk_size, num_tasks, num_calc) = map(int, sys.argv[1:])

    start = time()

    with ProcessPoolExecutor(max_workers=max_workers) as executor:
        params = map(lambda _: (random(), num_calc), range(num_tasks))
        results = executor.map(task, params, chunksize=chunk_size)
    getLogger().info(sum(results))

    getLogger().info("{:.3f}".format(time() - start))

Parameterbeschreibung

• max_workers
• Maximale Anzahl von Prozessen, die gleichzeitig ausgeführt werden
• chunk_size
• Anzahl der Aufgaben, die gleichzeitig an den Prozess übergeben werden sollen?
• num_tasks
• Anzahl der Aufgabenausführungen
• num_calc
• Anzahl der in einer Aufgabe durchgeführten Berechnungen

Ausführungsumgebung

• CPU:Intel Core i5(2.6 GHz)
• Anzahl der Kerne: 2

Ausführungsergebnis

Die oberen vier sind das Ergebnis der Ausführung einer großen Anzahl großer Aufgaben und die unteren vier sind das Ergebnis der Ausführung einer großen Anzahl kleiner Aufgaben. Es scheint, dass die Angabe von chunk_size wichtig ist, wenn eine große Anzahl kleiner Aufgaben ausgeführt wird.