Teilen viele queues unter-Prozesse in Python

Es klingt wie Sie Ihre Probleme begannen, als Sie versuchte eine multiprocessing.Queue() durch übergeben es als argument. Sie können dies umgehen, durch Schaffung einer verwaltete Warteschlange statt:

import multiprocessing
manager = multiprocessing.Manager()
passable_queue = manager.Queue()

Wenn Sie ein manager, um es zu schaffen, Sie sind die Speicherung und Weitergabe um eine proxy der Warteschlange, anstatt die queue selbst, so dass selbst wenn das Objekt, das Sie passieren, um Ihren worker-Prozesse ist kopiert, es wird immer noch an der gleichen zugrunde liegenden Daten-Struktur: die Warteschlange. Es ist sehr ähnlich wie (in-Konzept) zu Zeigern in C/C++. Wenn Sie erstellen Sie Ihre Warteschlangen diese Weise werden Sie in der Lage, Sie zu übergeben, wenn Sie starten Sie einen Arbeitsprozess.

Da können Sie Warteschlangen, um jetzt brauchen Sie nicht mehr über Ihr Wörterbuch verwaltet werden. Ein normales Wörterbuch in main gespeichert werden alle Zuordnungen, und geben nur Ihren worker-Prozesse, die Warteschlangen, die Sie benötigen, so dass Sie nicht brauchen Zugang zu allen Zuordnungen.

Ich geschrieben habe, ein Beispiel von diesem hier. Es sieht aus wie Sie sind übergeben von Objekten zwischen Ihrer Arbeitnehmer, so dass das, was getan hier. Sich vorstellen, wir haben zwei Stufen der Verarbeitung, und die Daten sowohl beginnt und endet bei der Kontrolle der main. Schauen Sie, wie können wir die Warteschlangen, die eine Verbindung der Arbeiter wie eine pipeline, sondern indem Sie Sie nur Warteschlangen, die Sie benötigen, es gibt keine Notwendigkeit für Sie zu wissen, über alle Zuweisungen:

import multiprocessing as mp

def stage1(q_in, q_out):

    q_out.put(q_in.get()+"Stage 1 did some work.\n")
    return

def stage2(q_in, q_out):

    q_out.put(q_in.get()+"Stage 2 did some work.\n")
    return

def main():

    pool = mp.Pool()
    manager = mp.Manager()

    # create managed queues
    q_main_to_s1 = manager.Queue()
    q_s1_to_s2 = manager.Queue()
    q_s2_to_main = manager.Queue()

    # launch workers, passing them the queues they need
    results_s1 = pool.apply_async(stage1, (q_main_to_s1, q_s1_to_s2))
    results_s2 = pool.apply_async(stage2, (q_s1_to_s2, q_s2_to_main))

    # Send a message into the pipeline
    q_main_to_s1.put("Main started the job.\n")

    # Wait for work to complete
    print(q_s2_to_main.get()+"Main finished the job.")

    pool.close()
    pool.join()

    return

if __name__ == "__main__":
    main()

Den code erzeugt diese Ausgabe:

Main mit der Arbeit anfingen.

Stufe 1 hat einige Arbeit.

Stufe 2 hat einige Arbeit.

Main beendet den job.

Ich nicht beinhalten ein Beispiel der Speicherung der Warteschlangen oder AsyncResults Objekte in Wörterbüchern, weil ich noch nicht ganz verstanden, wie dein Programm funktionieren soll. Aber jetzt, können Sie übergeben Sie Ihre queues frei, Sie können bauen Sie Ihr Wörterbuch zum speichern der Warteschlange/Prozess-mappings, wie gebraucht.

In der Tat, wenn Sie wirklich tun, bauen eine pipeline zwischen mehreren Arbeitnehmern, die Sie gar nicht benötigen, um einen Verweis auf die "inter-worker" - Warteschlangen in main. Erstellen von Warteschlangen, übergeben Sie an Ihre Arbeitnehmer, dann nur behalten, Verweise auf Warteschlangen main verwenden. Ich würde definitiv empfehlen, versuchen zu lassen alte Warteschlangen, Müll gesammelt, so schnell wie möglich, wenn Sie wirklich tun haben "eine beliebige Anzahl" von Warteschlangen.