RabbitMQ / AMQP: einzelne Warteschlange, mehrere Verbraucher für dieselbe Nachricht?

143

Ich fange gerade an, RabbitMQ und AMQP im Allgemeinen zu verwenden.

  • Ich habe eine Warteschlange mit Nachrichten
  • Ich habe mehrere Verbraucher, die ich mit derselben Nachricht verschiedene Dinge tun möchte .

Der größte Teil der RabbitMQ-Dokumentation scheint sich auf Round-Robin zu konzentrieren, dh wenn eine einzelne Nachricht von einem einzelnen Verbraucher konsumiert wird, wobei die Last auf jeden Verbraucher verteilt wird. Dies ist in der Tat das Verhalten, das ich sehe.

Ein Beispiel: Der Produzent hat eine einzelne Warteschlange und sendet alle 2 Sekunden Nachrichten:

var amqp = require('amqp');
var connection = amqp.createConnection({ host: "localhost", port: 5672 });
var count = 1;

connection.on('ready', function () {
  var sendMessage = function(connection, queue_name, payload) {
    var encoded_payload = JSON.stringify(payload);  
    connection.publish(queue_name, encoded_payload);
  }

  setInterval( function() {    
    var test_message = 'TEST '+count
    sendMessage(connection, "my_queue_name", test_message)  
    count += 1;
  }, 2000) 


})

Und hier ist ein Verbraucher:

var amqp = require('amqp');
var connection = amqp.createConnection({ host: "localhost", port: 5672 });
connection.on('ready', function () {
  connection.queue("my_queue_name", function(queue){
    queue.bind('#'); 
    queue.subscribe(function (message) {
      var encoded_payload = unescape(message.data)
      var payload = JSON.parse(encoded_payload)
      console.log('Recieved a message:')
      console.log(payload)
    })
  })
})

Wenn ich den Konsumenten zweimal starte, kann ich sehen, dass jeder Konsument alternative Nachrichten im Round-Robin-Verhalten konsumiert. ZB sehe ich die Nachrichten 1, 3, 5 in einem Terminal, 2, 4, 6 im anderen .

Meine Frage ist:

  • Kann ich jeden Verbraucher die gleichen Nachrichten empfangen lassen? Dh beide Verbraucher erhalten Nachricht 1, 2, 3, 4, 5, 6? Wie heißt das in AMQP / RabbitMQ? Wie ist es normalerweise konfiguriert?

  • Wird das allgemein gemacht? Sollte ich die Nachricht stattdessen nur durch den Austausch in zwei separate Warteschlangen mit einem einzigen Verbraucher weiterleiten lassen?

Mikemaccana
quelle
5
Ich bin kein RabbitMQ-Experte. Was Sie jetzt haben, heißt Warteschlange, aber was Sie wollen, sind Themen. Weitere Informationen finden Sie in diesem Tutorial: rabbitmq.com/tutorials/tutorial-five-python.html . Weitere Informationen zu Warteschlangen und Themen: msdn.microsoft.com/en-us /library/windowsazure/hh367516.aspx
UrbanEsc
1
Ich glaube, er will eigentlich Fanout, obwohl die Themen auch funktionieren und später mehr Kontrolle geben werden.
Robthewolf
Danke @UrbanEsc. Themen scheinen das Problem zu lösen, indem eine Nachricht mehrere Warteschlangen trifft und daher von den Verbrauchern jeder Warteschlange verwendet wird. Das lenkt mich weiter in Richtung des Szenarios mit mehreren Warteschlangen / einzelnen Verbrauchern für meinen speziellen Fall.
Mikemaccana
1
Für 2018 (und sogar für 2016 und früher) lautet die Antwort, etwas wie Kafka, IMO, zu verwenden.
WattsInABox

Antworten:

115

Kann ich jeden Verbraucher die gleichen Nachrichten empfangen lassen? Dh beide Verbraucher erhalten Nachricht 1, 2, 3, 4, 5, 6? Wie heißt das in AMQP / RabbitMQ? Wie ist es normalerweise konfiguriert?

Nein, nicht wenn sich die Verbraucher in derselben Warteschlange befinden. Aus dem AMQP Concepts Guide von RabbitMQ :

Es ist wichtig zu verstehen, dass in AMQP 0-9-1 Nachrichten zwischen Verbrauchern ausgetauscht werden.

Dies scheint zu implizieren, dass das Round-Robin-Verhalten innerhalb einer Warteschlange gegeben und nicht konfigurierbar ist. Das heißt, separate Warteschlangen sind erforderlich, damit dieselbe Nachrichten-ID von mehreren Verbrauchern verarbeitet werden kann.

Wird das allgemein gemacht? Sollte ich die Nachricht stattdessen nur durch den Austausch in zwei separate Warteschlangen mit einem einzigen Verbraucher weiterleiten lassen?

Nein, es ist nicht so, dass eine einzelne Warteschlange / mehrere Verbraucher, bei denen jeder Verbraucher dieselbe Nachrichten-ID verarbeitet, nicht möglich sind. Es ist in der Tat besser, wenn der Austausch die Nachricht in zwei separate Warteschlangen weiterleitet.

Da ich kein zu komplexes Routing benötige, wird ein Fanout-Austausch dies gut handhaben. Ich habe mich früher nicht zu sehr auf den Austausch konzentriert, da node-amqp das Konzept eines "Standardaustauschs" hat, mit dem Sie Nachrichten direkt in einer Verbindung veröffentlichen können. Die meisten AMQP-Nachrichten werden jedoch in einem bestimmten Austausch veröffentlicht.

Hier ist mein Fanout-Austausch, sowohl beim Senden als auch beim Empfangen:

var amqp = require('amqp');
var connection = amqp.createConnection({ host: "localhost", port: 5672 });
var count = 1;

connection.on('ready', function () {
  connection.exchange("my_exchange", options={type:'fanout'}, function(exchange) {   

    var sendMessage = function(exchange, payload) {
      console.log('about to publish')
      var encoded_payload = JSON.stringify(payload);
      exchange.publish('', encoded_payload, {})
    }

    // Recieve messages
    connection.queue("my_queue_name", function(queue){
      console.log('Created queue')
      queue.bind(exchange, ''); 
      queue.subscribe(function (message) {
        console.log('subscribed to queue')
        var encoded_payload = unescape(message.data)
        var payload = JSON.parse(encoded_payload)
        console.log('Recieved a message:')
        console.log(payload)
      })
    })

    setInterval( function() {    
      var test_message = 'TEST '+count
      sendMessage(exchange, test_message)  
      count += 1;
    }, 2000) 
 })
})
Mikemaccana
quelle
1
Fan Out war eindeutig das, was Sie wollten. Es würde Ihnen hier nicht helfen, aber ich dachte, ich würde erwähnen, dass das Round-Robin-Verhalten innerhalb einer Warteschlange konfigurierbar ist. int prefetchCount = 1; channel.basicQos(prefetchCount); Auf diese Weise kann jeder Verbraucher eine Nachricht erhalten, sobald die vorherige Nachricht fertig ist. Anstatt abwechselnde Nachrichten zu empfangen. Wiederum löst es Ihr Problem nicht, könnte aber nützlich sein, damit die Leute es wissen. Beispiel hier http://www.rabbitmq.com/tutorials/tutorial-two-java.html unter Fair Dispatch
Ommit
3
Zur Verdeutlichung: 'Standardaustausch' ist nicht knotenamqp-spezifisch. Es handelt sich um ein allgemeines AMQP-Konzept mit folgenden Regeln: Wenn eine Nachricht für den Standardaustausch veröffentlicht wird, wird der Routing-Schlüssel (mit dem diese Nachricht veröffentlicht wurde) vom AMQP-Broker als Warteschlangenname behandelt. Sie können also anscheinend direkt in Warteschlangen veröffentlichen. Aber du bist nicht. Der Broker bindet einfach jede Warteschlange an den Standardaustausch mit einem Routing-Schlüssel, der dem Namen der Warteschlange entspricht.
Ruslan Stelmachenko
1
Gibt es eine Alternative zu Apache activemq jms-Themen in rabbitmq, bei denen keine Warteschlangen beteiligt sind, sondern Multicasting?
Pantonis
Wenn sich derselbe Benutzer von mehreren Geräten aus anmeldet, erhält die Nachricht nur ein Gerät. Wie kann es gelöst werden oder eine Idee bitte?
Rafiq
@ Rafiq du solltest eine Frage dazu stellen.
Mikemaccana
27

Lesen Sie einfach das rabbitmq-Tutorial . Sie veröffentlichen Nachrichten zum Austauschen und nicht zum Anstehen. Es wird dann an die entsprechenden Warteschlangen weitergeleitet. In Ihrem Fall sollten Sie für jeden Verbraucher eine separate Warteschlange binden. Auf diese Weise können sie Nachrichten völlig unabhängig voneinander konsumieren.

Driushkin
quelle
24

Die letzten paar Antworten sind fast richtig - ich habe Unmengen von Apps, die Nachrichten generieren, die bei verschiedenen Verbrauchern landen müssen, sodass der Vorgang sehr einfach ist.

Wenn Sie möchten, dass mehrere Verbraucher dieselbe Nachricht erhalten, gehen Sie wie folgt vor.

Erstellen Sie mehrere Warteschlangen, eine für jede App, die die Nachricht empfangen soll. Binden Sie in jeder Warteschlangeneigenschaft ein Routing-Tag an den amq.direct-Austausch. Ändern Sie Ihre Veröffentlichungs-App so, dass sie an amq.direct gesendet wird, und verwenden Sie das Routing-Tag (keine Warteschlange). AMQP kopiert dann die Nachricht mit derselben Bindung in jede Warteschlange. Klappt wunderbar :)

Beispiel: Nehmen wir an, ich habe eine JSON-Zeichenfolge, die ich generiere. Ich veröffentliche sie mit dem Routing-Tag "new-sales-order" an der Vermittlungsstelle "amq.direct". Ich habe eine Warteschlange für meine order_printer-App, die die Bestellung druckt. Ich habe eine Warteschlange für mein Abrechnungssystem, das eine Kopie der Bestellung sendet und dem Kunden eine Rechnung stellt. Ich habe ein Webarchivsystem, in dem ich Bestellungen aus historischen / Compliance-Gründen archiviere, und eine Weboberfläche des Kunden, in der Bestellungen nachverfolgt werden, wenn andere Informationen eingehen eine Bestellung.

Meine Warteschlangen sind also: order_printer, order_billing, order_archive und order_tracking. Alle haben das Bindungs-Tag "new-sales-order", alle 4 erhalten die JSON-Daten.

Dies ist eine ideale Möglichkeit, Daten zu senden, ohne dass die Veröffentlichungs-App die empfangenden Apps kennt oder sich darum kümmert.

z900collector
quelle
7

Ja, jeder Verbraucher kann dieselben Nachrichten empfangen. Besuchen Sie http://www.rabbitmq.com/tutorials/tutorial-three-python.html http://www.rabbitmq.com/tutorials/tutorial-four-python.html http: //www.rabbitmq. com / tutorials / tutorial-five-python.html

für verschiedene Möglichkeiten zum Weiterleiten von Nachrichten. Ich weiß, dass sie für Python und Java sind, aber es ist gut, die Prinzipien zu verstehen, zu entscheiden, was Sie tun, und dann in JS zu finden, wie es geht. Es hört sich so an, als ob Sie ein einfaches Fanout ( Tutorial 3 ) durchführen möchten , das die Nachrichten an alle mit dem Austausch verbundenen Warteschlangen sendet.

Der Unterschied zu dem, was Sie tun und was Sie tun möchten, besteht im Wesentlichen darin, dass Sie Fanout einrichten und austauschen oder eingeben. Fanout-Excahnges senden alle Nachrichten an alle verbundenen Warteschlangen. Jede Warteschlange hat einen Verbraucher, der separat auf alle Nachrichten zugreifen kann.

Ja, dies wird üblicherweise durchgeführt. Dies ist eine der Funktionen von AMPQ.

Robthewolf
quelle
gute Antwort, außer durch "wird das gewöhnlich gemacht?" Ich bezog mich auf "jeden Verbraucher die gleichen Nachrichten empfangen lassen" - was normalerweise nicht der Fall ist (Verbraucher in derselben Warteschlange immer Round Robin). Wahrscheinlich meine Schuld daran, dass ich nicht klar genug bin.
Mikemaccana
Eigentlich würde ich sagen, dass es davon abhängt, wofür Sie es verwenden möchten. Sie haben zwei grundlegende Möglichkeiten: Pub / Sub oder Arbeitswarteschlangen. Ihre ursprüngliche Einrichtung war eine Arbeitswarteschlange, aber was Sie wollten, war ein Fanout-Pub / Sub. Sie weisen darauf hin, dass die allgemeine Verwendung hier völlig davon abhängt, was Sie tun möchten.
Robthewolf
Sicher, aber in einer Arbeitswarteschlange wird dieselbe Nachricht (z. B. dieselbe Nachrichten-ID) nicht von verschiedenen Verbrauchern verarbeitet - es handelt sich implizit um Round Robin. Auch dies ist wahrscheinlich meine Schuld daran, dass ich nicht klar genug bin.
Mikemaccana
wir scheinen hier über Kreuz zu reden.
Robthewolf
Entschuldigung für die Verwirrung. Wenn es eine Möglichkeit gibt, eine Arbeitswarteschlange zu haben, in der Verbraucher in derselben Warteschlange dieselbe Nachrichten-ID verarbeiten, verweisen Sie mich bitte auf eine Referenz. Ansonsten werde ich weiterhin glauben, was ich woanders gelesen habe.
Mikemaccana
3

RabbitMQ / AMQP: einzelne Warteschlange, mehrere Verbraucher für dieselbe Nachricht und Seitenaktualisierung.

rabbit.on('ready', function () {    });
    sockjs_chat.on('connection', function (conn) {

        conn.on('data', function (message) {
            try {
                var obj = JSON.parse(message.replace(/\r/g, '').replace(/\n/g, ''));

                if (obj.header == "register") {

                    // Connect to RabbitMQ
                    try {
                        conn.exchange = rabbit.exchange(exchange, { type: 'topic',
                            autoDelete: false,
                            durable: false,
                            exclusive: false,
                            confirm: true
                        });

                        conn.q = rabbit.queue('my-queue-'+obj.agentID, {
                            durable: false,
                            autoDelete: false,
                            exclusive: false
                        }, function () {
                            conn.channel = 'my-queue-'+obj.agentID;
                            conn.q.bind(conn.exchange, conn.channel);

                            conn.q.subscribe(function (message) {
                                console.log("[MSG] ---> " + JSON.stringify(message));
                                conn.write(JSON.stringify(message) + "\n");
                            }).addCallback(function(ok) {
                                ctag[conn.channel] = ok.consumerTag; });
                        });
                    } catch (err) {
                        console.log("Could not create connection to RabbitMQ. \nStack trace -->" + err.stack);
                    }

                } else if (obj.header == "typing") {

                    var reply = {
                        type: 'chatMsg',
                        msg: utils.escp(obj.msga),
                        visitorNick: obj.channel,
                        customField1: '',
                        time: utils.getDateTime(),
                        channel: obj.channel
                    };

                    conn.exchange.publish('my-queue-'+obj.agentID, reply);
                }

            } catch (err) {
                console.log("ERROR ----> " + err.stack);
            }
        });

        // When the visitor closes or reloads a page we need to unbind from RabbitMQ?
        conn.on('close', function () {
            try {

                // Close the socket
                conn.close();

                // Close RabbitMQ           
               conn.q.unsubscribe(ctag[conn.channel]);

            } catch (er) {
                console.log(":::::::: EXCEPTION SOCKJS (ON-CLOSE) ::::::::>>>>>>> " + er.stack);
            }
        });
    });
Durai
quelle
1

Lassen Sie einfach jeden Verbraucher aus seiner eigenen Warteschlange konsumieren, um das gewünschte Verhalten zu erzielen. Sie müssen einen nicht direkten Austauschtyp (Thema, Header, Fanout) verwenden, um die Nachricht gleichzeitig an alle Warteschlangen zu senden.

Skylos
quelle
1

Wie ich Ihren Fall einschätze, ist:

  • Ich habe eine Warteschlange mit Nachrichten (Ihre Quelle für den Empfang von Nachrichten, nennen wir sie q111)

  • Ich habe mehrere Verbraucher, die ich mit derselben Nachricht verschiedene Dinge tun möchte.

Ihr Problem hierbei ist, dass während 3 Nachrichten von dieser Warteschlange empfangen werden, Nachricht 1 von einem Verbraucher A verbraucht wird, andere Verbraucher B und C die Nachricht 2 und 3 verbrauchen. Wenn Sie ein Setup benötigen, bei dem rabbitmq dieselben Kopien von weiterleitet alle diese drei Nachrichten (1,2,3) an alle drei verbundenen Verbraucher (A, B, C) gleichzeitig.

Um dies zu erreichen, können viele Konfigurationen vorgenommen werden. Eine einfache Möglichkeit besteht darin, das folgende zweistufige Konzept zu verwenden:

  • Verwenden Sie eine dynamische rabbitmq-Schaufel, um Nachrichten aus der gewünschten Warteschlange (q111) aufzunehmen und in einem Fanout-Austausch zu veröffentlichen (Austausch, der ausschließlich für diesen Zweck erstellt und reserviert wurde).
  • Konfigurieren Sie nun Ihre Kunden A, B & C (die die Warteschlange abgehört haben (q111)) so, dass sie von diesem Fanout-Austausch direkt mit einer exklusiven und anonymen Warteschlange für jeden Verbraucher abhören.

Hinweis: Verwenden Sie dieses Konzept nicht direkt aus der Quellwarteschlange (q111), da bereits verbrauchte Nachrichten nicht an Ihren Fanout-Austausch gesendet werden.

Wenn Sie der Meinung sind, dass dies nicht Ihren genauen Anforderungen entspricht, können Sie gerne Ihre Vorschläge posten :-)

Anshuman Banerjee
quelle
0

Ich denke, Sie sollten das Senden Ihrer Nachrichten mit dem Fan-Out- Austauscher überprüfen . Auf diese Weise erhalten Sie dieselbe Nachricht für verschiedene Verbraucher. Unter der Tabelle erstellt RabbitMQ für jeden dieser neuen Verbraucher / Abonnenten unterschiedliche Warteschlangen.

Dies ist der Link, um das Tutorial-Beispiel in Javascript https://www.rabbitmq.com/tutorials/tutorial-one-javascript.html zu sehen

Alejandro Serret
quelle
-1

In diesem Szenario gibt es eine interessante Option, die ich in den Antworten hier nicht gefunden habe.

Sie können Nachrichten mit der Funktion "Request" in einem Consumer nacken, um sie in einem anderen zu verarbeiten. Im Allgemeinen ist es kein richtiger Weg, aber vielleicht ist es gut genug für jemanden.

https://www.rabbitmq.com/nack.html

Und hüte dich vor Schleifen (wenn alle Verbraucher nackt sind + Nachricht anfordern)!

Alexus1024
quelle
1
Ich würde sehr davon abraten, da es in keiner Weise skaliert. Es gibt keine Bestellung für Verbraucher. Sie können nicht garantieren, dass Verbraucher B, der sie nicht anfordert, die Nachricht vor Verbraucher A empfängt, der sie verarbeitet und anfordert. Die genannten Schleifen sind ein Problem. Wie Sie sagen "das ist im Allgemeinen nicht der richtige Weg", und ich kann mir kein Szenario vorstellen, in dem dies besser wäre als die anderen Antworten.
Kevin Streicher