RabbitMQ: persistente Nachricht mit Themenaustausch

Ich bin ganz neu RabbitMQ.

Ich habe ein 'Thema' exchange. Der Verbraucher kann gestartet werden, nachdem der publisher. Ich möchte die Verbraucher in der Lage sein, Nachrichten zu empfangen, die gesendet wurden, bevor Sie wurden, und das waren noch nicht verbraucht.

Dem exchange eingerichtet ist, mit den folgenden Parametern:

exchange_type => 'topic'
durable => 1
auto_delete => 0
passive => 0

Den Nachrichten veröffentlicht werden, mit diesem parameter:

delivery_mode => 2

Verbraucher verwenden Sie get() zum abrufen der Nachrichten von der Börse.

Leider, werden alle Nachrichten veröffentlicht, bevor irgendein Kunde war verloren. Ich habe es mit verschiedenen Kombinationen.

Ich vermute, mein problem ist, dass der Austausch nicht halten Nachrichten. Vielleicht brauche ich ein queue zwischen dem Verlag und der Warteschlange. Aber das scheint nicht zu funktionieren, mit einem "topic" wechseln, wo Nachrichten weitergeleitet werden, die von einem Schlüssel.

Irgendeine Idee, wie ich Vorgehen sollte. Ich benutze das Perl-binding Net::RabbitMQ (sollte egal sein) und RabbitMQ 2.2.0.

InformationsquelleAutor der Frage Julien | 2011-05-27

  1. 57

    Benötigen Sie eine dauerhafte Warteschlange zum speichern von Nachrichten, wenn es keine angeschlossene Verbraucher zum verarbeiten der Nachrichten, die zum Zeitpunkt der Veröffentlichung.

    Einem exchange speichert keine Nachrichten, sondern eine Warteschlange. Die verwirrende Teil ist, dass der Austausch gekennzeichnet werden kann als "robust", aber das wirklich bedeutet, ist, dass die exchange selbst wird noch da sein, wenn Sie starten Sie Ihren broker, aber es hat nicht bedeuten, dass alle Nachrichten gesendet, die exchange automatisch beibehalten.

    Da, hier sind zwei Optionen:

    1. Durchführen einen administrativen Schritt bevor Sie beginnen Ihre Publisher zu erstellen, die Warteschlange(N) selbst. Könnten Sie die web-Benutzeroberfläche oder die Befehlszeile Werkzeuge, dies zu tun. Stellen Sie sicher, schaffen es wie eine dauerhafte queue, so dass es speichert alle Nachrichten, die weitergeleitet werden, selbst wenn keine aktiven Verbraucher.
    2. Vorausgesetzt die Konsumenten sind so codiert, dass immer erklären (und damit auto-create) den Austausch und die Warteschlangen beim Start (und dass Sie erklären, dass Sie Sie als dauerhaft), nur alle deine Verbraucher mindestens einmal vor Beginn eines jeden publishers. Das wird sicherstellen, dass alle Ihre Warteschlangen erstellt werden richtig. Sie können dann Herunterfahren der Verbraucher, wenn Sie wirklich nötig sind, weil die Warteschlangen persistieren alle zukünftigen Nachrichten an Sie weitergeleitet.

    Würde ich für #1. Es kann nicht sein, viele Schritte zu vollziehen und Sie konnte immer Skript die erforderlichen Schritte, so dass Sie wiederholt werden könnte. Plus, wenn Sie alle Ihre Verbraucher gehen, um zu ziehen aus der gleichen queue (anstatt einer dedizierten Warteschlange jeder) es ist wirklich ein minimales Stück von Verwaltungsaufwand.

    Warteschlangen sind etwas verwaltet und gesteuert werden, richtig. Andernfalls könnten Sie am Ende mit rogue Verbraucher deklarieren dauerhafte Warteschlangen, Sie für ein paar Minuten, aber nie wieder. Bald nach Sie haben eine ständig wachsende Warteschlange mit nichts reduziert seine Größe, und einer drohenden broker Apokalypse.

    InformationsquelleAutor der Antwort Brian Kelly

  2. 16

    Wie erwähnt von Brian exchange speichert keine Nachrichten, und ist hauptsächlich verantwortlich für die Weiterleitung von Nachrichten an ein anderes exchange/s oder queue/s. Wenn der Austausch ist nicht gebunden an eine Warteschlange, dann werden alle Nachrichten gesendet, die exchange 'verloren'

    Sollte man nicht erklären müssen behoben client-Warteschlangen in der publisher-Skript aus, da diese möglicherweise nicht skalierbar. Warteschlangen können erstellt werden, die dynamisch durch Ihren Verleger und weitergeleitet, die intern mithilfe des exchange-zu-exchange-Bindung.

    RabbitMQ unterstützt exchange-zu-exchange-Bindungen, ermöglichen die Topologie Flexibilität, die Entkopplung und andere Vorteile. Lesen Sie hier mehr zu RabbitMQ von Exchange auf Exchange-Bindungen [AMPQ]

    RabbitMQ Von Exchange Auf Exchange-Bindung

    RabbitMQ: persistente Nachricht mit Themenaustausch

    Beispiel Python-code zu erstellen, die exchange-zu-exchange-verbindlich, mit Ausdauer, wenn kein Verbraucher vorhanden ist, mit Warteschlange. 

    #!/usr/bin/env python
import pika
import sys


connection = pika.BlockingConnection(pika.ConnectionParameters(
host='localhost'))
channel = connection.channel()


#Declares the entry exchange to be used by all producers to send messages. Could be external producers as well
channel.exchange_declare(exchange='data_gateway',
exchange_type='fanout',
durable=True,
auto_delete=False)

#Declares the processing exchange to be used.Routes messages to various queues. For internal use only
channel.exchange_declare(exchange='data_distributor',
exchange_type='topic',
durable=True,
auto_delete=False)

#Binds the external/producer facing exchange to the internal exchange
channel.exchange_bind(destination='data_distributor',source='data_gateway')

##Create Durable Queues binded to the data_distributor exchange
channel.queue_declare(queue='trade_db',durable=True)
channel.queue_declare(queue='trade_stream_service',durable=True)
channel.queue_declare(queue='ticker_db',durable=True)
channel.queue_declare(queue='ticker_stream_service',durable=True)
channel.queue_declare(queue='orderbook_db',durable=True)
channel.queue_declare(queue='orderbook_stream_service',durable=True)

#Bind queues to exchanges and correct routing key. Allows for messages to be saved when no consumer is present
channel.queue_bind(queue='orderbook_db',exchange='data_distributor',routing_key='*.*.orderbook')
channel.queue_bind(queue='orderbook_stream_service',exchange='data_distributor',routing_key='*.*.orderbook')
channel.queue_bind(queue='ticker_db',exchange='data_distributor',routing_key='*.*.ticker')
channel.queue_bind(queue='ticker_stream_service',exchange='data_distributor',routing_key='*.*.ticker')
channel.queue_bind(queue='trade_db',exchange='data_distributor',routing_key='*.*.trade')
channel.queue_bind(queue='trade_stream_service',exchange='data_distributor',routing_key='*.*.trade')

    InformationsquelleAutor der Antwort Skillachie

Schreibe einen Kommentar