Warteschlange remote-Anrufe auf einer Python-Twisted-Perspektive broker?

Einer ready-made-option, die helfen könnte, mit dieser ist twisted.internet.defer.DeferredSemaphore. Dies ist die asynchrone version der normalen (zählen) der semaphore, die Sie vielleicht bereits wissen, wenn Sie haben viel getan, threaded Programmierung.

A (zu zählen) ein semaphor ist ein viel wie ein mutex (eine Sperre). Aber wo ein mutex kann nur einmal erworben werden, bis eine entsprechende Freigabe ein (zählen), semaphore kann so konfiguriert werden, erlauben eine beliebige (aber festgelegte) Anzahl von Akquisitionen erfolgreich zu sein, bevor die entsprechenden Veröffentlichungen sind erforderlich.

Hier ist ein Beispiel für die Verwendung DeferredSemaphore laufen zehn asynchrone Operationen, aber die laufen bei den meisten drei von Ihnen auf einmal:

from twisted.internet.defer import DeferredSemaphore, gatherResults
from twisted.internet.task import deferLater
from twisted.internet import reactor


def async(n):
    print 'Starting job', n
    d = deferLater(reactor, n, lambda: None)
    def cbFinished(ignored):
        print 'Finishing job', n
    d.addCallback(cbFinished)
    return d


def main():
    sem = DeferredSemaphore(3)

    jobs = []
    for i in range(10):
        jobs.append(sem.run(async, i))

    d = gatherResults(jobs)
    d.addCallback(lambda ignored: reactor.stop())
    reactor.run()


if __name__ == '__main__':
    main()

DeferredSemaphore auch explizit acquire und release Methoden, aber die run Methode ist so bequem, es ist fast immer was Sie wollen. Es ruft die acquire Methode, die gibt einen Deferred. Den ersten Deferred es fügt einen callback, welche die Funktion aufruft, Sie übergeben (ggf. zusammen mit positions-oder keyword-Argumenten). Wenn diese Funktion gibt eine Deferred, dann das zweite Deferred wird ein callback Hinzugefügt, die fordert, die release Methode.

Synchronen Fall wird behandelt, indem sofort aufrufen release. Fehler werden auch behandelt, indem Sie propagieren, sondern dafür, dass die notwendigen release getan wird, lassen Sie die DeferredSemaphore in einem konsistenten Zustand. Das Ergebnis der Funktion übergeben run (oder das Ergebnis der Deferred es gibt) wird das Ergebnis der Deferred zurückgegeben run.

Ein weiterer möglicher Ansatz könnte sein, basierend auf DeferredQueue und cooperate. DeferredQueue ist meist wie eine normale Schlange, aber seine get Methode gibt eine Deferred. Wenn es passiert, werden keine Elemente in der queue, die zur Zeit des Aufrufs, die Deferred wird nicht ausgelöst, bis ein Element Hinzugefügt wird.

Hier ein Beispiel:

from random import randrange

from twisted.internet.defer import DeferredQueue
from twisted.internet.task import deferLater, cooperate
from twisted.internet import reactor


def async(n):
    print 'Starting job', n
    d = deferLater(reactor, n, lambda: None)
    def cbFinished(ignored):
        print 'Finishing job', n
    d.addCallback(cbFinished)
    return d


def assign(jobs):
    # Create new jobs to be processed
    jobs.put(randrange(10))
    reactor.callLater(randrange(10), assign, jobs)


def worker(jobs):
    while True:
        yield jobs.get().addCallback(async)


def main():
    jobs = DeferredQueue()

    for i in range(10):
        jobs.put(i)

    assign(jobs)

    for i in range(3):
        cooperate(worker(jobs))

    reactor.run()


if __name__ == '__main__':
    main()

Beachten Sie, dass die async worker-Funktion ist die gleiche wie die aus dem ersten Beispiel. Dieses mal jedoch, es gibt auch eine worker Funktion, die explizit ziehen Arbeitsplätze aus der DeferredQueue Bearbeiten mit async (durch hinzufügen async als ein Rückruf der Deferred zurückgegeben get). Die worker generator angetrieben durch cooperate, denen geht es einmal nach jedem Deferred es ergibt sich feuert. Die main-loop dann startet diese drei worker-Generatoren, so dass drei Arbeitsplätze in progress zu einem bestimmten Zeitpunkt.

Dieser Ansatz beinhaltet, dass ein bisschen mehr code, als die DeferredSemaphore Ansatz, hat aber einige Vorteile, die interessant sein können. Zuerst cooperate gibt eine CooperativeTask Instanz, die nützliche Methoden, wie pause, resume, und ein paar andere. Außerdem werden alle Aufträge zugewiesen, die gleiche Mitarbeiterin, wird kooperieren mit einander in Planung, so dass nicht zu überlasten die Ereignis-Schleife (und dies ist, was gibt die API auch seinen Namen). Auf der DeferredQueue Seite, ist es auch möglich, Grenzen zu setzen, wie viele Artikel sich derzeit in Bearbeitung, so können Sie vermeiden, vollständig überlastung Ihrer server (zum Beispiel, wenn Sie Ihre Bild-Prozessoren stecken und stop, Erledigung von Aufgaben). Wenn der code ruft put verarbeitet die queue-überlauf-Ausnahme, Sie können verwenden Sie diese als Druck, um zu versuchen zu stoppen, keine neuen Arbeitsplätze (vielleicht rangieren Sie auf einem anderen server, oder Alarmierung ein administrator). Ähnliche Dinge tut mit DeferredSemaphore ist ein bisschen schwieriger, da es keine Möglichkeit zu begrenzen, wie viele Arbeitsplätze warten zu können, erwerben die semaphore.