Wie die Anzahl der gleichzeitigen Threads in Python zu begrenzen?

Question

Wie kann ich die Anzahl der gleichzeitigen Threads in Python begrenzen?

Zum Beispiel habe ich ein Verzeichnis mit vielen Dateien, und ich möchte alle von ihnen, aber nur 4 gleichzeitig parallel verarbeiten.

Hier ist, was ich bisher:

def process_file(fname): 
     # open file and do something                        

def process_file_thread(queue, fname): 
    queue.put(process_file(fname)) 

def process_all_files(d): 
    files=glob.glob(d + '/*') 
    q=Queue.Queue() 
    for fname in files: 
     t=threading.Thread(target=process_file_thread, args=(q, fname)) 
     t.start() 
    q.join() 

def main(): 
    process_all_files('.') 
    # Do something after all files have been processed 

Wie kann ich den Code ändern, so dass nur 4 Threads gleichzeitig ausgeführt werden?

Beachten Sie, dass ich warten möchte, bis alle Dateien verarbeitet wurden, und fahre dann mit den verarbeiteten Dateien fort.

Answer 1

Zum Beispiel habe ich ein Verzeichnis mit vielen Dateien, und ich möchte alle von ihnen, aber nur 4 gleichzeitig parallel verarbeiten.

Das ist genau das, was ein Thread-Pool tut: Sie erstellen Jobs, und der Pool läuft 4 gleichzeitig parallel. Sie können die Dinge noch einfacher machen, indem Sie einen Executor verwenden, wo Sie ihm einfach Funktionen (oder andere Callables) übergeben und Ihnen Futures für die Ergebnisse zurückgeben. Sie können all dies selbst erstellen, aber Sie müssen nicht. *

Das concurrent.futures Modul der Stdlib ist der einfachste Weg, dies zu tun. (Für Python 3.1 und früher, siehe backport.) Tatsächlich ist one of the main examples sehr nah an dem, was Sie tun möchten. Aber es uns genau auf Ihre Anwendungsfall lassen sich anpassen:

def process_all_files(d): 
    files = glob.glob(d + '/*') 
    with concurrent.futures.ThreadPoolExecutor(max_workers=4) as executor: 
     fs = [executor.submit(process_file, file) for file in files] 
     concurrent.futures.wait(fs) 

Wenn Sie process_file wollte etwas zurückgeben, das ist fast so einfach:

def process_all_files(d): 
    files = glob.glob(d + '/*') 
    with concurrent.futures.ThreadPoolExecutor(max_workers=4) as executor: 
     fs = [executor.submit(process_file, file) for file in files] 
     for f in concurrent.futures.as_completed(fs): 
      do_something(f.result()) 

Und wenn Sie wollen auch Ausnahmen behandeln ... na ja, nur schauen im Beispiel; es ist nur ein try/except um den Anruf zu result().

---

* Wenn Sie sie selbst bauen wollen, ist es nicht so schwer. Die Quelle zu multiprocessing.pool ist gut geschrieben und kommentiert, und nicht so kompliziert, und die meisten harten Sachen sind nicht relevant für Threading; Die Quelle zu ist noch einfacher.

Answer 2

ich diese Technik ein paar Mal verwendet, ich glaube, es ist ein bisschen hässlich Gedanke ist:

import threading 

def process_something(): 
    something = list(get_something) 

    def worker(): 
     while something: 
      obj = something.pop() 
      # do something with obj 

    threads = [Thread(target=worker) for i in range(4)] 
    [t.start() for t in threads] 
    [t.join() for t in threads]