Wie kann ich Threading in Python verwenden?

Ich versuche, das Threading in Python zu verstehen. Ich habe mir die Dokumentation und Beispiele angesehen, aber ehrlich gesagt sind viele Beispiele zu raffiniert und ich habe Probleme, sie zu verstehen.

Wie zeigen Sie deutlich, dass Aufgaben für Multithreading aufgeteilt sind?

Seit diese Frage im Jahr 2010 gestellt wurde, wurde das einfache Multithreading mit Python mit Map und Pool erheblich vereinfacht .

Der folgende Code stammt aus einem Artikel / Blog-Beitrag, den Sie unbedingt lesen sollten (keine Zugehörigkeit) - Parallelität in einer Zeile: Ein besseres Modell für alltägliche Threading-Aufgaben . Ich fasse unten zusammen - es sind nur ein paar Codezeilen:

from multiprocessing.dummy import Pool as ThreadPool
pool = ThreadPool(4)
results = pool.map(my_function, my_array)

Welches ist die Multithread-Version von:

results = []
for item in my_array:
    results.append(my_function(item))

Beschreibung

Map ist eine coole kleine Funktion und der Schlüssel zum einfachen Einfügen von Parallelität in Ihren Python-Code. Für Unbekannte ist Map etwas, das aus funktionalen Sprachen wie Lisp übernommen wurde. Es ist eine Funktion, die eine andere Funktion über eine Sequenz abbildet.

Map übernimmt für uns die Iteration über die Sequenz, wendet die Funktion an und speichert alle Ergebnisse am Ende in einer praktischen Liste.

Implementierung

Parallele Versionen der Kartenfunktion werden von zwei Bibliotheken bereitgestellt: Multiprocessing und das wenig bekannte, aber ebenso fantastische Stiefkind: multiprocessing.dummy.

multiprocessing.dummyist genau das gleiche wie das Multiprocessing-Modul, verwendet jedoch stattdessen Threads ( ein wichtiger Unterschied - verwenden Sie mehrere Prozesse für CPU-intensive Aufgaben; Threads für (und während) E / A ):

multiprocessing.dummy repliziert die API von multiprocessing, ist jedoch nur ein Wrapper um das Threading-Modul.

import urllib2
from multiprocessing.dummy import Pool as ThreadPool

urls = [
  'http://www.python.org',
  'http://www.python.org/about/',
  'http://www.onlamp.com/pub/a/python/2003/04/17/metaclasses.html',
  'http://www.python.org/doc/',
  'http://www.python.org/download/',
  'http://www.python.org/getit/',
  'http://www.python.org/community/',
  'https://wiki.python.org/moin/',
]

# Make the Pool of workers
pool = ThreadPool(4)

# Open the URLs in their own threads
# and return the results
results = pool.map(urllib2.urlopen, urls)

# Close the pool and wait for the work to finish
pool.close()
pool.join()

Und das Timing ergibt sich:

Single thread:   14.4 seconds
       4 Pool:   3.1 seconds
       8 Pool:   1.4 seconds
      13 Pool:   1.3 seconds

Übergeben mehrerer Argumente (funktioniert nur in Python 3.3 und höher so ):

So übergeben Sie mehrere Arrays:

results = pool.starmap(function, zip(list_a, list_b))

Oder um eine Konstante und ein Array zu übergeben:

results = pool.starmap(function, zip(itertools.repeat(constant), list_a))

Wenn Sie eine frühere Version von Python verwenden, können Sie über diese Problemumgehung mehrere Argumente übergeben .

(Danke an user136036 für den hilfreichen Kommentar.)

Hier ist ein einfaches Beispiel: Sie müssen einige alternative URLs ausprobieren und den Inhalt der ersten, die antwortet, zurückgeben.

import Queue
import threading
import urllib2

# Called by each thread
def get_url(q, url):
    q.put(urllib2.urlopen(url).read())

theurls = ["http://google.com", "http://yahoo.com"]

q = Queue.Queue()

for u in theurls:
    t = threading.Thread(target=get_url, args = (q,u))
    t.daemon = True
    t.start()

s = q.get()
print s

Dies ist ein Fall, in dem Threading als einfache Optimierung verwendet wird: Jeder Subthread wartet darauf, dass eine URL aufgelöst und beantwortet wird, um seinen Inhalt in die Warteschlange zu stellen. Jeder Thread ist ein Daemon (hält den Prozess nicht aufrecht, wenn der Hauptthread endet - das ist häufiger als nicht). Der Hauptthread startet alle Subthreads, führt getin der Warteschlange ein, um zu warten, bis einer von ihnen a ausgeführt hat put, gibt dann die Ergebnisse aus und wird beendet (wodurch alle Subthreads entfernt werden, die möglicherweise noch ausgeführt werden, da es sich um Daemon-Threads handelt).

Die ordnungsgemäße Verwendung von Threads in Python ist ausnahmslos mit E / A-Vorgängen verbunden (da CPython ohnehin nicht mehrere Kerne zum Ausführen von CPU-gebundenen Aufgaben verwendet, besteht der einzige Grund für das Threading darin, den Prozess nicht zu blockieren, während auf einige E / A gewartet wird ). Warteschlangen sind übrigens fast immer der beste Weg, um Arbeit an Threads zu verteilen und / oder die Arbeitsergebnisse zu sammeln. Sie sind an sich threadsicher, sodass Sie sich keine Gedanken über Sperren, Bedingungen, Ereignisse, Semaphoren und andere Interaktionen machen müssen -Thread-Koordinations- / Kommunikationskonzepte.

ANMERKUNG : Für die eigentliche Parallelisierung in Python sollten Sie das Multiprocessing- Modul verwenden, um mehrere Prozesse zu verzweigen, die parallel ausgeführt werden (aufgrund der globalen Interpretersperre bieten Python-Threads Interleaving, werden jedoch tatsächlich seriell, nicht parallel und nur ausgeführt nützlich beim Verschachteln von E / A-Operationen).

Wenn Sie jedoch nur nach Verschachtelung suchen (oder E / A-Operationen ausführen, die trotz der globalen Interpretersperre parallelisiert werden können), ist das Threading- Modul der Ausgangspunkt. Betrachten wir als wirklich einfaches Beispiel das Problem der Summierung eines großen Bereichs durch parallele Summierung von Unterbereichen:

import threading

class SummingThread(threading.Thread):
     def __init__(self,low,high):
         super(SummingThread, self).__init__()
         self.low=low
         self.high=high
         self.total=0

     def run(self):
         for i in range(self.low,self.high):
             self.total+=i


thread1 = SummingThread(0,500000)
thread2 = SummingThread(500000,1000000)
thread1.start() # This actually causes the thread to run
thread2.start()
thread1.join()  # This waits until the thread has completed
thread2.join()
# At this point, both threads have completed
result = thread1.total + thread2.total
print result

Beachten Sie, dass das obige Beispiel ein sehr dummes Beispiel ist, da es absolut keine E / A-Vorgänge ausführt und seriell ausgeführt wird, obwohl es aufgrund der globalen Interpretersperre in CPython verschachtelt ist (mit dem zusätzlichen Aufwand für die Kontextumschaltung) .

Wie bereits erwähnt, kann CPython Threads nur für E / A-Wartezeiten aufgrund von GIL verwenden .

Wenn Sie für CPU-gebundene Aufgaben von mehreren Kernen profitieren möchten, verwenden Sie Multiprocessing :

from multiprocessing import Process

def f(name):
    print 'hello', name

if __name__ == '__main__':
    p = Process(target=f, args=('bob',))
    p.start()
    p.join()

Nur ein Hinweis: Für das Threading ist keine Warteschlange erforderlich.

Dies ist das einfachste Beispiel, das ich mir vorstellen kann und das zeigt, dass 10 Prozesse gleichzeitig ausgeführt werden.

import threading
from random import randint
from time import sleep


def print_number(number):

    # Sleeps a random 1 to 10 seconds
    rand_int_var = randint(1, 10)
    sleep(rand_int_var)
    print "Thread " + str(number) + " slept for " + str(rand_int_var) + " seconds"

thread_list = []

for i in range(1, 10):

    # Instantiates the thread
    # (i) does not make a sequence, so (i,)
    t = threading.Thread(target=print_number, args=(i,))
    # Sticks the thread in a list so that it remains accessible
    thread_list.append(t)

# Starts threads
for thread in thread_list:
    thread.start()

# This blocks the calling thread until the thread whose join() method is called is terminated.
# From http://docs.python.org/2/library/threading.html#thread-objects
for thread in thread_list:
    thread.join()

# Demonstrates that the main process waited for threads to complete
print "Done"

Die Antwort von Alex Martelli hat mir geholfen. Hier ist jedoch eine modifizierte Version, die ich (zumindest für mich) für nützlicher hielt.

Aktualisiert: Funktioniert sowohl in Python 2 als auch in Python 3

try:
    # For Python 3
    import queue
    from urllib.request import urlopen
except:
    # For Python 2 
    import Queue as queue
    from urllib2 import urlopen

import threading

worker_data = ['http://google.com', 'http://yahoo.com', 'http://bing.com']

# Load up a queue with your data. This will handle locking
q = queue.Queue()
for url in worker_data:
    q.put(url)

# Define a worker function
def worker(url_queue):
    queue_full = True
    while queue_full:
        try:
            # Get your data off the queue, and do some work
            url = url_queue.get(False)
            data = urlopen(url).read()
            print(len(data))

        except queue.Empty:
            queue_full = False

# Create as many threads as you want
thread_count = 5
for i in range(thread_count):
    t = threading.Thread(target=worker, args = (q,))
    t.start()

Fädeln Sie eine gegebene Funktion folgendermaßen ein f:

import threading
threading.Thread(target=f).start()

Argumente an übergeben f

threading.Thread(target=f, args=(a,b,c)).start()

Ich fand das sehr nützlich: Erstellen Sie so viele Threads wie Kerne und lassen Sie sie eine (große) Anzahl von Aufgaben ausführen (in diesem Fall ein Shell-Programm aufrufen):

import Queue
import threading
import multiprocessing
import subprocess

q = Queue.Queue()
for i in range(30): # Put 30 tasks in the queue
    q.put(i)

def worker():
    while True:
        item = q.get()
        # Execute a task: call a shell program and wait until it completes
        subprocess.call("echo " + str(item), shell=True)
        q.task_done()

cpus = multiprocessing.cpu_count() # Detect number of cores
print("Creating %d threads" % cpus)
for i in range(cpus):
     t = threading.Thread(target=worker)
     t.daemon = True
     t.start()

q.join() # Block until all tasks are done

Python 3 bietet die Möglichkeit, parallele Aufgaben zu starten . Dies erleichtert unsere Arbeit.

Es verfügt über Thread-Pooling und Prozess-Pooling .

Folgendes gibt einen Einblick:

ThreadPoolExecutor Beispiel ( Quelle )

import concurrent.futures
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/']

# Retrieve a single page and report the URL and contents
def load_url(url, timeout):
    with urllib.request.urlopen(url, timeout=timeout) as conn:
        return conn.read()

# We can use a with statement to ensure threads are cleaned up promptly
with concurrent.futures.ThreadPoolExecutor(max_workers=5) as executor:
    # Start the load operations and mark each future with its URL
    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)))

ProcessPoolExecutor ( Quelle )

import concurrent.futures
import math

PRIMES = [
    112272535095293,
    112582705942171,
    112272535095293,
    115280095190773,
    115797848077099,
    1099726899285419]

def is_prime(n):
    if n % 2 == 0:
        return False

    sqrt_n = int(math.floor(math.sqrt(n)))
    for i in range(3, sqrt_n + 1, 2):
        if n % i == 0:
            return False
    return True

def main():
    with concurrent.futures.ProcessPoolExecutor() as executor:
        for number, prime in zip(PRIMES, executor.map(is_prime, PRIMES)):
            print('%d is prime: %s' % (number, prime))

if __name__ == '__main__':
    main()

Verwenden des brandneuen Moduls concurrent.futures

def sqr(val):
    import time
    time.sleep(0.1)
    return val * val

def process_result(result):
    print(result)

def process_these_asap(tasks):
    import concurrent.futures

    with concurrent.futures.ProcessPoolExecutor() as executor:
        futures = []
        for task in tasks:
            futures.append(executor.submit(sqr, task))

        for future in concurrent.futures.as_completed(futures):
            process_result(future.result())
        # Or instead of all this just do:
        # results = executor.map(sqr, tasks)
        # list(map(process_result, results))

def main():
    tasks = list(range(10))
    print('Processing {} tasks'.format(len(tasks)))
    process_these_asap(tasks)
    print('Done')
    return 0

if __name__ == '__main__':
    import sys
    sys.exit(main())

Der Executor-Ansatz scheint allen bekannt zu sein, die sich zuvor mit Java die Hände schmutzig gemacht haben.

Nebenbei bemerkt: Um das Universum gesund zu halten, vergessen Sie nicht, Ihre Pools / Executoren zu schließen, wenn Sie keinen withKontext verwenden (was so großartig ist, dass es das für Sie tut).

Für mich ist das perfekte Beispiel für das Threading die Überwachung asynchroner Ereignisse. Schauen Sie sich diesen Code an.

# thread_test.py
import threading
import time

class Monitor(threading.Thread):
    def __init__(self, mon):
        threading.Thread.__init__(self)
        self.mon = mon

    def run(self):
        while True:
            if self.mon[0] == 2:
                print "Mon = 2"
                self.mon[0] = 3;

Sie können mit diesem Code spielen, indem Sie eine IPython- Sitzung öffnen und Folgendes tun:

>>> from thread_test import Monitor
>>> a = [0]
>>> mon = Monitor(a)
>>> mon.start()
>>> a[0] = 2
Mon = 2
>>>a[0] = 2
Mon = 2

Warte ein paar Minuten

>>> a[0] = 2
Mon = 2

Die meisten Dokumentationen und Tutorials verwenden Pythons Threadingund QueueModule und können für Anfänger überwältigend erscheinen.

Betrachten Sie vielleicht das concurrent.futures.ThreadPoolExecutorModul von Python 3.

In Kombination mit withKlausel- und Listenverständnis könnte dies ein echter Reiz sein.

from concurrent.futures import ThreadPoolExecutor, as_completed

def get_url(url):
    # Your actual program here. Using threading.Lock() if necessary
    return ""

# List of URLs to fetch
urls = ["url1", "url2"]

with ThreadPoolExecutor(max_workers = 5) as executor:

    # Create threads
    futures = {executor.submit(get_url, url) for url in urls}

    # as_completed() gives you the threads once finished
    for f in as_completed(futures):
        # Get the results
        rs = f.result()

Ich habe hier viele Beispiele gesehen, bei denen keine wirkliche Arbeit ausgeführt wurde und die größtenteils an die CPU gebunden waren. Hier ist ein Beispiel für eine CPU-gebundene Aufgabe, die alle Primzahlen zwischen 10 und 10,05 Millionen berechnet. Ich habe hier alle vier Methoden angewendet:

import math
import timeit
import threading
import multiprocessing
from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor


def time_stuff(fn):
    """
    Measure time of execution of a function
    """
    def wrapper(*args, **kwargs):
        t0 = timeit.default_timer()
        fn(*args, **kwargs)
        t1 = timeit.default_timer()
        print("{} seconds".format(t1 - t0))
    return wrapper

def find_primes_in(nmin, nmax):
    """
    Compute a list of prime numbers between the given minimum and maximum arguments
    """
    primes = []

    # Loop from minimum to maximum
    for current in range(nmin, nmax + 1):

        # Take the square root of the current number
        sqrt_n = int(math.sqrt(current))
        found = False

        # Check if the any number from 2 to the square root + 1 divides the current numnber under consideration
        for number in range(2, sqrt_n + 1):

            # If divisible we have found a factor, hence this is not a prime number, lets move to the next one
            if current % number == 0:
                found = True
                break

        # If not divisible, add this number to the list of primes that we have found so far
        if not found:
            primes.append(current)

    # I am merely printing the length of the array containing all the primes, but feel free to do what you want
    print(len(primes))

@time_stuff
def sequential_prime_finder(nmin, nmax):
    """
    Use the main process and main thread to compute everything in this case
    """
    find_primes_in(nmin, nmax)

@time_stuff
def threading_prime_finder(nmin, nmax):
    """
    If the minimum is 1000 and the maximum is 2000 and we have four workers,
    1000 - 1250 to worker 1
    1250 - 1500 to worker 2
    1500 - 1750 to worker 3
    1750 - 2000 to worker 4
    so let’s split the minimum and maximum values according to the number of workers
    """
    nrange = nmax - nmin
    threads = []
    for i in range(8):
        start = int(nmin + i * nrange/8)
        end = int(nmin + (i + 1) * nrange/8)

        # Start the thread with the minimum and maximum split up to compute
        # Parallel computation will not work here due to the GIL since this is a CPU-bound task
        t = threading.Thread(target = find_primes_in, args = (start, end))
        threads.append(t)
        t.start()

    # Don’t forget to wait for the threads to finish
    for t in threads:
        t.join()

@time_stuff
def processing_prime_finder(nmin, nmax):
    """
    Split the minimum, maximum interval similar to the threading method above, but use processes this time
    """
    nrange = nmax - nmin
    processes = []
    for i in range(8):
        start = int(nmin + i * nrange/8)
        end = int(nmin + (i + 1) * nrange/8)
        p = multiprocessing.Process(target = find_primes_in, args = (start, end))
        processes.append(p)
        p.start()

    for p in processes:
        p.join()

@time_stuff
def thread_executor_prime_finder(nmin, nmax):
    """
    Split the min max interval similar to the threading method, but use a thread pool executor this time.
    This method is slightly faster than using pure threading as the pools manage threads more efficiently.
    This method is still slow due to the GIL limitations since we are doing a CPU-bound task.
    """
    nrange = nmax - nmin
    with ThreadPoolExecutor(max_workers = 8) as e:
        for i in range(8):
            start = int(nmin + i * nrange/8)
            end = int(nmin + (i + 1) * nrange/8)
            e.submit(find_primes_in, start, end)

@time_stuff
def process_executor_prime_finder(nmin, nmax):
    """
    Split the min max interval similar to the threading method, but use the process pool executor.
    This is the fastest method recorded so far as it manages process efficiently + overcomes GIL limitations.
    RECOMMENDED METHOD FOR CPU-BOUND TASKS
    """
    nrange = nmax - nmin
    with ProcessPoolExecutor(max_workers = 8) as e:
        for i in range(8):
            start = int(nmin + i * nrange/8)
            end = int(nmin + (i + 1) * nrange/8)
            e.submit(find_primes_in, start, end)

def main():
    nmin = int(1e7)
    nmax = int(1.05e7)
    print("Sequential Prime Finder Starting")
    sequential_prime_finder(nmin, nmax)
    print("Threading Prime Finder Starting")
    threading_prime_finder(nmin, nmax)
    print("Processing Prime Finder Starting")
    processing_prime_finder(nmin, nmax)
    print("Thread Executor Prime Finder Starting")
    thread_executor_prime_finder(nmin, nmax)
    print("Process Executor Finder Starting")
    process_executor_prime_finder(nmin, nmax)

main()

Hier sind die Ergebnisse auf meinem Mac OS X-Vierkerncomputer

Sequential Prime Finder Starting
9.708213827005238 seconds
Threading Prime Finder Starting
9.81836523200036 seconds
Processing Prime Finder Starting
3.2467174359990167 seconds
Thread Executor Prime Finder Starting
10.228896902000997 seconds
Process Executor Finder Starting
2.656402041000547 seconds

Hier ist das sehr einfache Beispiel für den CSV- Import mithilfe von Threading. (Die Aufnahme in die Bibliothek kann für verschiedene Zwecke unterschiedlich sein.)

Hilfsfunktionen:

from threading import Thread
from project import app
import csv


def import_handler(csv_file_name):
    thr = Thread(target=dump_async_csv_data, args=[csv_file_name])
    thr.start()

def dump_async_csv_data(csv_file_name):
    with app.app_context():
        with open(csv_file_name) as File:
            reader = csv.DictReader(File)
            for row in reader:
                # DB operation/query

Treiberfunktion:

import_handler(csv_file_name)

Ich möchte mit einem einfachen Beispiel und den Erklärungen, die ich nützlich fand, als ich dieses Problem selbst angehen musste, einen Beitrag leisten.

In dieser Antwort finden Sie einige Informationen zu Pythons GIL (Global Interpreter Lock) und ein einfaches Beispiel aus dem Alltag, das mit multiprocessing.dummy geschrieben wurde, sowie einige einfache Benchmarks.

Global Interpreter Lock (GIL)

Python erlaubt kein Multithreading im wahrsten Sinne des Wortes. Es verfügt über ein Multithreading-Paket. Wenn Sie jedoch ein Multithreading-Paket verwenden möchten, um Ihren Code zu beschleunigen, ist es normalerweise keine gute Idee, es zu verwenden.

Python hat ein Konstrukt namens Global Interpreter Lock (GIL). Die GIL stellt sicher, dass immer nur einer Ihrer 'Threads' gleichzeitig ausgeführt werden kann. Ein Thread erwirbt die GIL, erledigt ein wenig Arbeit und leitet die GIL dann an den nächsten Thread weiter.

Dies geschieht sehr schnell, so dass es für das menschliche Auge so aussieht, als würden Ihre Threads parallel ausgeführt, aber sie wechseln sich nur mit demselben CPU-Kern ab.

All diese GIL-Übergaben erhöhen den Aufwand für die Ausführung. Dies bedeutet, dass die Verwendung des Threading-Pakets häufig keine gute Idee ist, wenn Sie Ihren Code schneller ausführen möchten.

Es gibt Gründe, das Threading-Paket von Python zu verwenden. Wenn Sie einige Dinge gleichzeitig ausführen möchten und Effizienz kein Problem darstellt, ist dies völlig in Ordnung und praktisch. Oder wenn Sie Code ausführen, der auf etwas warten muss (wie z. B. einige E / A), kann dies sehr sinnvoll sein. In der Threading-Bibliothek können Sie jedoch keine zusätzlichen CPU-Kerne verwenden.

Multithreading kann an das Betriebssystem ausgelagert werden (durch Multiverarbeitung) und an eine externe Anwendung, die Ihren Python-Code aufruft (z. B. Spark oder Hadoop ), oder an einen Code, den Ihr Python-Code aufruft (z. B.: Sie könnten Lassen Sie Ihren Python-Code eine C-Funktion aufrufen, die die teuren Multithread-Aufgaben erledigt.

Warum das wichtig ist

Weil viele Leute viel Zeit damit verbringen, Engpässe in ihrem ausgefallenen Python-Multithread-Code zu finden, bevor sie lernen, was die GIL ist.

Sobald diese Informationen klar sind, ist hier mein Code:

#!/bin/python
from multiprocessing.dummy import Pool
from subprocess import PIPE,Popen
import time
import os

# In the variable pool_size we define the "parallelness".
# For CPU-bound tasks, it doesn't make sense to create more Pool processes
# than you have cores to run them on.
#
# On the other hand, if you are using I/O-bound tasks, it may make sense
# to create a quite a few more Pool processes than cores, since the processes
# will probably spend most their time blocked (waiting for I/O to complete).
pool_size = 8

def do_ping(ip):
    if os.name == 'nt':
        print ("Using Windows Ping to " + ip)
        proc = Popen(['ping', ip], stdout=PIPE)
        return proc.communicate()[0]
    else:
        print ("Using Linux / Unix Ping to " + ip)
        proc = Popen(['ping', ip, '-c', '4'], stdout=PIPE)
        return proc.communicate()[0]


os.system('cls' if os.name=='nt' else 'clear')
print ("Running using threads\n")
start_time = time.time()
pool = Pool(pool_size)
website_names = ["www.google.com","www.facebook.com","www.pinterest.com","www.microsoft.com"]
result = {}
for website_name in website_names:
    result[website_name] = pool.apply_async(do_ping, args=(website_name,))
pool.close()
pool.join()
print ("\n--- Execution took {} seconds ---".format((time.time() - start_time)))

# Now we do the same without threading, just to compare time
print ("\nRunning NOT using threads\n")
start_time = time.time()
for website_name in website_names:
    do_ping(website_name)
print ("\n--- Execution took {} seconds ---".format((time.time() - start_time)))

# Here's one way to print the final output from the threads
output = {}
for key, value in result.items():
    output[key] = value.get()
print ("\nOutput aggregated in a Dictionary:")
print (output)
print ("\n")

print ("\nPretty printed output: ")
for key, value in output.items():
    print (key + "\n")
    print (value)

Hier ist Multithreading mit einem einfachen Beispiel, das hilfreich sein wird. Sie können es ausführen und leicht verstehen, wie Multithreading in Python funktioniert. Ich habe eine Sperre verwendet, um den Zugriff auf andere Threads zu verhindern, bis die vorherigen Threads ihre Arbeit beendet haben. Durch die Verwendung dieser Codezeile

tLock = threading.BoundedSemaphore (Wert = 4)

Sie können mehrere Prozesse gleichzeitig zulassen und den Rest der Threads beibehalten, die später oder nach Abschluss vorheriger Prozesse ausgeführt werden.

import threading
import time

#tLock = threading.Lock()
tLock = threading.BoundedSemaphore(value=4)
def timer(name, delay, repeat):
    print  "\r\nTimer: ", name, " Started"
    tLock.acquire()
    print "\r\n", name, " has the acquired the lock"
    while repeat > 0:
        time.sleep(delay)
        print "\r\n", name, ": ", str(time.ctime(time.time()))
        repeat -= 1

    print "\r\n", name, " is releaseing the lock"
    tLock.release()
    print "\r\nTimer: ", name, " Completed"

def Main():
    t1 = threading.Thread(target=timer, args=("Timer1", 2, 5))
    t2 = threading.Thread(target=timer, args=("Timer2", 3, 5))
    t3 = threading.Thread(target=timer, args=("Timer3", 4, 5))
    t4 = threading.Thread(target=timer, args=("Timer4", 5, 5))
    t5 = threading.Thread(target=timer, args=("Timer5", 0.1, 5))

    t1.start()
    t2.start()
    t3.start()
    t4.start()
    t5.start()

    print "\r\nMain Complete"

if __name__ == "__main__":
    Main()

Wenn wir uns diesen Beitrag ausleihen, wissen wir, wie man zwischen Multithreading, Multiprocessing und Async / asynciound deren Verwendung wählt .

Python 3 verfügt über eine neue integrierte Bibliothek für Parallelität und Parallelität: concurrent.futures

Also werde ich durch ein Experiment demonstrieren, wie man vier Aufgaben (dh eine .sleep()Methode) auf folgende Threading-PoolWeise ausführt:

from concurrent.futures import ThreadPoolExecutor, as_completed
from time import sleep, time

def concurrent(max_worker=1):
    futures = []

    tick = time()
    with ThreadPoolExecutor(max_workers=max_worker) as executor:
        futures.append(executor.submit(sleep, 2))  # Two seconds sleep
        futures.append(executor.submit(sleep, 1))
        futures.append(executor.submit(sleep, 7))
        futures.append(executor.submit(sleep, 3))

        for future in as_completed(futures):
            if future.result() is not None:
                print(future.result())

    print('Total elapsed time by {} workers:'.format(max_worker), time()-tick)

concurrent(5)
concurrent(4)
concurrent(3)
concurrent(2)
concurrent(1)

Ausgabe:

Total elapsed time by 5 workers: 7.007831811904907
Total elapsed time by 4 workers: 7.007944107055664
Total elapsed time by 3 workers: 7.003149509429932
Total elapsed time by 2 workers: 8.004627466201782
Total elapsed time by 1 workers: 13.013478994369507

[ HINWEIS ]:

• Wie Sie in den obigen Ergebnissen sehen können, waren 3 Mitarbeiter für diese vier Aufgaben der beste Fall .
• Wenn Sie einen Prozess Aufgabe statt Ich habe / O gebunden oder Sperrung ( multiprocessingvs threading) Sie das ändern könnte ThreadPoolExecutorzu ProcessPoolExecutor.

Keine der vorherigen Lösungen verwendete tatsächlich mehrere Kerne auf meinem GNU / Linux-Server (wo ich keine Administratorrechte habe). Sie liefen nur auf einem einzigen Kern.

Ich habe die os.forkSchnittstelle der unteren Ebene verwendet, um mehrere Prozesse zu erzeugen. Dies ist der Code, der für mich funktioniert hat:

from os import fork

values = ['different', 'values', 'for', 'threads']

for i in range(len(values)):
    p = fork()
    if p == 0:
        my_function(values[i])
        break

import threading
import requests

def send():

  r = requests.get('https://www.stackoverlow.com')

thread = []
t = threading.Thread(target=send())
thread.append(t)
t.start()