Rabbitmq: Ändern Sie Nachrichtentext vor dem Verzehr

Question

Ich bin neu in RabbitMQ und ich möchte eine Nachricht vor dem Konsumieren von einer Warteschlange ändern. Ich habe einen Austausch, der unantastbar bleiben sollte. Der Client empfängt Nachrichten mit einem bestimmten Routing-Schlüssel. Aber da sind viele von ihnen und ich möchte den Körper filtern und ändern, bevor ich sie in die Warteschlange verlese.

Exchange-Nachrichten bevölkern wie folgt aussieht:

{ 
"_context_domain": "unsuitable", 
"_msg_id": "1", 
"_context_quota_class": null, 
"_context_read_only": false, 
"_context_request_id": "1" 
} 

{ 
"_context_domain": "suitable", 
"_msg_id": "2", 
"_context_quota_class": null, 
"_context_read_only": false, 
"_context_request_id": "2" 
} 

Gibt es eine Möglichkeit, sie zu filtern und zu modifizieren, bevor raubend? Zum Beispiel:

... 
channel.queueBind(QUEUE_NAME, "EXCHANGE_NAME", "ROUTNG_KEY"); 
final Consumer consumer = new DefaultConsumer(channel) { 
     @Override 
     public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { 
       String message = new String(body, "UTF-8"); 
       Gson mapper = new Gson(); 
       SomeObject object = (SomeObject) mapper.fromJson(message, SomeObject.class); 
        if (SomeObject.getContext_domain = "suitable"){ 
          //publish somehow SomeObject.getMsg_id into QUEUE_NAME 
} 
} 

Gibt es eine Möglichkeit, es zu tun?

Answer 1

Von AMQP unterstützte Austauschvorgänge erlauben keine Filterung des Nachrichtentexts. Sie könnten "topic" or "header" exchanges verwenden, die Nachrichten basierend auf dem Routing-Schlüssel oder einer Nachrichtenkopfzeile weiterleiten können.

Keine von ihnen erlaubt Ihnen jedoch, die Nachricht selbst zu ändern. Wenn Sie das tun möchten, müssen Sie Ihr eigenes RabbitMQ-Plugin entwickeln, das einen Austausch implementiert.

Wenn Sie diese Straße gehen wollen, ist es ziemlich einfach zu tun. Allerdings sind Börsen nicht genau für diesen Zweck ausgelegt: Sie sind einfach Routing-Tabellen. Wenn Sie das tun, wird der Channel-Prozess in RabbitMQ Ressourcen verbrauchen, um die gewünschten Änderungen vorzunehmen (dh ein Austausch ist kein Prozess, sondern im Prinzip eine Zeile in einer Tabelle). Das bedeutet auch, dass andere Clients auf derselben Verbindung blockiert werden können, bis der Kanal fertig ist, um eine Nachricht zu ändern und in eine Warteschlange zu stellen.

Ein gängigerer Weg, um zu erreichen, was Sie wollen, ist die Verwendung eines Consumers, um alle Nachrichten zu verarbeiten, die dort benötigten Filterungen/Modifikationen durchzuführen und das Ergebnis in eine zweite Warteschlange einzureihen. Diese zweite Warteschlange würde von Ihren normalen Arbeitern verbraucht werden.