Multiplexing RabbitMQ-Nachricht

Question

Zum Beispiel habe ich 4 Quellen, die Metriken veröffentlichen. Ich möchte alle diese Nachrichten in einer Warteschlange / einem Austausch multiplexen / zusammenführen

--------+----+----+----+----+        -------+---------+----+---------+---------+
Source1 | M1 | M2 | M3 |    |   =>   Result | M1 | M4 | M2 | M3 | M6 | M5 | M7 |
Source2 | M4 |    |    | M5 |
Source3 |    |    | M6 |    |
Source4 |    |    |    | M7 |

For each queue:
 * Read one message
 * Publish message to the Result queue

Gibt es eine "native" Möglichkeit, dies in RabbitMQ zu tun, oder sollte ich meinen eigenen Consumer / Publisher schreiben?

BEARBEITEN 1

Ein Beispiel zur Verdeutlichung, sagen wir mal nach einiger Zeit

     Processing "window"
        +-+
Source1 |X|XXXXXXXXXXXXX
Source2 |Y|YYYYYYY
Source3 |Z|ZZZZZZZZZZ
Source4 |W|WW
        +-+

Und dann später

     Processing "window"
           +-+
Source1 XXX|X|XXXXXXXXXX
Source2 YYY|Y|YYYY
Source3 ZZZ|Z|ZZZZZZZ
Source4 WWW| |
           +-+

Und dann später

     Processing "window"
                 +-+
Source1 XXXXXXXXX|X|XXXX
Source2 YYYYYYYY | |
Source3 ZZZZZZZZZ|Z|Z
Source4 WWW      | |
                 +-+

Die ergebnisintensive Bestellung lautet: X Y Z W X Y Z W X Y Z W X Y Z X Y Z X Y Z X Y Z X Y Z X Z X Z X Z X X X

X, Y, Z, W dann X, Y, Z, W dann X, Y, Z, W dann X, Y, Z ... X, Z ...

Auf diese Weise können alle anderen Nachrichten aus anderen Quellen konsumiert werden, selbst wenn eine Quelle "spammt".

Aus technischen / finanziellen Gründen muss ich jeweils nur eine Nachricht konsumieren.

Der Verbraucher ist viel langsamer als die Produzenten, aber die Produzenten veröffentlichen viel, aber gelegentlich.

Wenn jede Quelle an einer Börse veröffentlicht wird, die an dieselbe Warteschlange gebunden ist, kann das Ergebnis XXXXXXXXXXXXXX YYYYYYYY ZZZZZZZZZZZ WWWoder sein XXXXX Y XXXXX YYY XXX YYYY ZZZZZZZZZZZ WWW(abhängig von der Veröffentlichungsrate jeder Quelle).

Answer 1

Ich denke, was Sie wollen, können Sie einfach erreichen, indem Sie ein einziges Skript ausführen, das alle Warteschlangen abonniert.

Die Hauptanforderung besteht darin, einen einzelnen Anwendungsthread zu verwenden, um alle Nachrichten zu verarbeiten, unabhängig davon, aus welcher Warteschlange sie kommen. Was das aussieht wie auf variieren je nachdem , welche Sprache und Client - Bibliothek Sie verwenden - wenn Sie PHP verwenden, die Sie wirklich von Ihrem Weg zu gehen haben würde nicht Single-Thread zu sein, aber vielleicht gibt es einige Client - Bibliotheken sind Dies setzt voraus, dass sich jeder Rückruf in einem separaten Arbeitsthread befindet und Sie eine gemeinsam genutzte Ressource benötigen, damit sie blockiert werden können.

In Bezug auf die tatsächliche RabbitMQ-Seite der Dinge müssen Sie:

• Registrieren Sie ein Abonnement für den Server, um Nachrichten an Sie zu senden basic.consume. Dies wird generell empfohlen, wenn basic.getohnehin explizit abgefragt wird
• Verwenden Sie für alle basic.consumeAnrufe einen einzigen "Kanal"
• Verwenden Sie manuelle Bestätigungen, damit Nachrichten in der Warteschlange bleiben, bis Ihr Prozess abgeschlossen ist
• Legen Sie ein Prefetch-Limit pro Warteschlange von 1 mit festbasic.qos

Wenn Sie 4 Warteschlangen haben, A, B, C und D, die beim Starten des Verbrauchers unterschiedlich viele Nachrichten enthalten:

1. Wenn Sie sich zum ersten Mal anmelden, bedeutet das Prefetch-Limit, dass eine Nachricht aus jeder Warteschlange an den Kanal gesendet wird. Nennen Sie sie A1, B1, C1 und D1
2. Die Clientbibliothek löst nacheinander für jedes dieser Ereignisse ein asynchrones Ereignis in Ihrer Anwendung aus
3. Ihr einzelner Worker-Thread behandelt das erste dieser Ereignisse und beginnt mit der Verarbeitung der Nachricht A1
4. Bis Sie diese Nachricht manuell bestätigen, können keine anderen Nachrichten eintreffen
5. Sobald Sie die erste Nachricht (A1) bestätigt haben, kann eine neue Nachricht aus dieser Warteschlange (A2) vorab abgerufen werden.
6. In der Zwischenzeit wird Ihr Worker-Thread das nächste Ereignis, das bereits ausgelöst wurde, für die Nachricht B1 entsperren und verarbeiten
7. Erst wenn Sie die ausstehenden Ereignisse für B1, C1 und D1 verarbeitet haben, wird dem Worker-Thread das Ereignis für die Nachricht A2 angezeigt
8. Solange in den Warteschlangen Nachrichten warten, werden diese im Round-Robin-Verfahren verarbeitet. Selbst wenn alle Warteschlangen bis auf eine leer werden, werden sie sofort nach Eingang einer Nachricht wieder in Rotation versetzt, da nur eine Nachricht aus der besetzten Warteschlange vorab abgerufen wurde. Der Rest wartet nur auf dem RabbitMQ-Server.