Rabbitmq：消費する前に本文を変更する

Question

私はRabbitMQで新しく、キューでそれを消費する前にメッセージを修正したいと思います。私は触れることができないはずの交換をしています。クライアントは、特定のルーティングキーを使用してメッセージを受信します。しかし、そこにはたくさんのものがあり、キューにそれらを公開する前にボディをフィルタリングして変更したいと思っています。

Exchangeがメッセージを取り込むには、次のようになります。

{ 
"_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" 
} 

がかかる前にそれらをフィルタリングし、修正する方法はありますか？たとえば ：

... 
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 
} 
} 

はそれを行うための方法はありますか？

Answer 1

AMQPでサポートされている交換は、メッセージ本文のフィルタリングを許可しません。ルーティングキーまたはメッセージヘッダーに基づいてメッセージをルーティングできる"topic" or "header" exchangesを使用できます。

ただし、メッセージ自体を変更することはできません。そうしたい場合は、交換を実装する独自のRabbitMQプラグインを開発する必要があります。

あなたがその道に行きたいなら、それはとても簡単です。しかし、エクスチェンジは正確にはこの目的のために設計されたものではなく、単にルーティングテーブルです。これを行うと、RabbitMQのチャネルプロセスは、必要な変更を行うためにリソースを消費します（つまり、交換はプロセスではなく、基本的にテーブルの行です）。つまり、同じ接続上の他のクライアントが、メッセージの修正とキュー登録を完了するまでブロックされる可能性があります。

消費者がすべてのメッセージを処理し、そこに必要なフィルタリングや変更を行い、その結果を第2のキューに入れることが、より一般的な方法です。この2番目の待ち行列は、通常の作業者によって消費されます。