Python und RabbitMQ - Beste Weg, um hören zu konsumieren Ereignisse, die von mehreren Kanälen?

Die Antwort auf die Frage "was ist der beste Weg" ist, hängt stark von Ihrem Nutzungsverhalten von Warteschlangen und was meinen Sie mit "am besten". Da kann ich nicht kommentieren, Fragen noch, ich werde nur versuchen, Ihnen einige mögliche Lösungen.

In jedem Beispiel werde ich davon ausgehen, Austausch ist bereits deklariert.

Threads

Können Sie verbrauchen die Nachrichten von zwei Warteschlangen, die auf separaten hosts in einzelnen Prozess mit pika.

Du hast Recht - als seine eigene FAQ Staaten, pika ist nicht thread-sicher, aber es kann verwendet werden, in multi-threaded-Weise durch die Schaffung von verbindungen zu RabbitMQ-hosts pro thread. Machen Sie dieses Beispiel laufen in threads mit threading Modul sieht wie folgt aus:

import pika
import threading


class ConsumerThread(threading.Thread):
    def __init__(self, host, *args, **kwargs):
        super(ConsumerThread, self).__init__(*args, **kwargs)

        self._host = host

    # Not necessarily a method.
    def callback_func(self, channel, method, properties, body):
        print("{} received '{}'".format(self.name, body))

    def run(self):
        credentials = pika.PlainCredentials("guest", "guest")

        connection = pika.BlockingConnection(
            pika.ConnectionParameters(host=self._host,
                                      credentials=credentials))

        channel = connection.channel()

        result = channel.queue_declare(exclusive=True)

        channel.queue_bind(result.method.queue,
                           exchange="my-exchange",
                           routing_key="*.*.*.*.*")

        channel.basic_consume(self.callback_func,
                              result.method.queue,
                              no_ack=True)

        channel.start_consuming()


if __name__ == "__main__":
    threads = [ConsumerThread("host1"), ConsumerThread("host2")]
    for thread in threads:
        thread.start()

Habe ich erklärt callback_func als eine Methode, rein zu verwenden ConsumerThread.name beim drucken message-body. Es könnte genauso gut eine Funktion außerhalb der ConsumerThread Klasse.

Prozesse

Alternativ können Sie immer nur das ausführen eines Prozesses mit consumer-code pro Warteschlange, die Sie konsumieren wollen events.

import pika
import sys


def callback_func(channel, method, properties, body):
    print(body)


if __name__ == "__main__":
    credentials = pika.PlainCredentials("guest", "guest")

    connection = pika.BlockingConnection(
        pika.ConnectionParameters(host=sys.argv[1],
                                  credentials=credentials))

    channel = connection.channel()

    result = channel.queue_declare(exclusive=True)

    channel.queue_bind(result.method.queue,
                       exchange="my-exchange",
                       routing_key="*.*.*.*.*")

    channel.basic_consume(callback_func, result.method.queue, no_ack=True)

    channel.start_consuming()

und dann ausführen von:

$ python single_consume.py host1
$ python single_consume.py host2  # e.g. on another console

Wenn die Arbeit, die Sie tun auf Nachrichten aus Warteschlangen ist CPU-heavy und solange die Anzahl der Kerne in Ihrer CPU >= Zahl der Verbraucher, es ist in der Regel besser für die Verwendung dieses Ansatzes - es sei denn, Ihrer Warteschlangen sind leer, die meisten von der Zeit und dem Verbraucher nicht nutzen, CPU-Zeit*.

Async

Andere alternative ist es, beinhalten einige asynchrone framework (zum Beispiel Verdreht) und läuft die ganze Sache im single-thread.

Können Sie nicht mehr nutzen BlockingConnection im asynchronen code; glücklicherweise pika hat adapter für Twisted:

from pika.adapters.twisted_connection import TwistedProtocolConnection
from pika.connection import ConnectionParameters
from twisted.internet import protocol, reactor, task
from twisted.python import log


class Consumer(object):
    def on_connected(self, connection):
        d = connection.channel()
        d.addCallback(self.got_channel)
        d.addCallback(self.queue_declared)
        d.addCallback(self.queue_bound)
        d.addCallback(self.handle_deliveries)
        d.addErrback(log.err)

    def got_channel(self, channel):
        self.channel = channel

        return self.channel.queue_declare(exclusive=True)

    def queue_declared(self, queue):
        self._queue_name = queue.method.queue

        self.channel.queue_bind(queue=self._queue_name,
                                exchange="my-exchange",
                                routing_key="*.*.*.*.*")

    def queue_bound(self, ignored):
        return self.channel.basic_consume(queue=self._queue_name)

    def handle_deliveries(self, queue_and_consumer_tag):
        queue, consumer_tag = queue_and_consumer_tag
        self.looping_call = task.LoopingCall(self.consume_from_queue, queue)

        return self.looping_call.start(0)

    def consume_from_queue(self, queue):
        d = queue.get()

        return d.addCallback(lambda result: self.handle_payload(*result))

    def handle_payload(self, channel, method, properties, body):
        print(body)


if __name__ == "__main__":
    consumer1 = Consumer()
    consumer2 = Consumer()

    parameters = ConnectionParameters()
    cc = protocol.ClientCreator(reactor,
                                TwistedProtocolConnection,
                                parameters)
    d1 = cc.connectTCP("host1", 5672)
    d1.addCallback(lambda protocol: protocol.ready)
    d1.addCallback(consumer1.on_connected)
    d1.addErrback(log.err)

    d2 = cc.connectTCP("host2", 5672)
    d2.addCallback(lambda protocol: protocol.ready)
    d2.addCallback(consumer2.on_connected)
    d2.addErrback(log.err)

    reactor.run()

Dieser Ansatz wäre sogar noch besser, je mehr Warteschlangen, die Sie verbrauchen würde und die weniger CPU-gebunden sind, die arbeiten ausführen, die von Verbrauchern ist*.

Python 3

Da Sie erwähnt haben pika habe ich eingeschränkt, mir zu Python 2.x-basierten Lösungen, weil pika ist noch nicht portiert.

Aber in Fall, dass Sie verschieben wollen, zu >=3.3, eine mögliche option ist die Verwendung asyncio mit einem AMQP-Protokoll (das Protokoll, die Sie sprechen mit RabbitMQ) , z.B. asynqp oder aioamqp.

* - bitte beachten Sie, dass diese sind sehr flach - Tipps- in den meisten Fällen ist die Wahl nicht so offensichtlich; was wird für Sie die beste ist, hängt von Warteschlangen "Sättigung" (Nachrichten/die Zeit) was tun Sie nach Erhalt dieser Nachrichten, in welcher Umgebung Sie führen den Verbraucher in etc.; es gibt keinen Weg, um sicher zu sein, als benchmark alle Implementierungen

InformationsquelleAutor Unit03