問題タブ [spring-amqp]

私は、RabbitMQ のやや特殊なユース ケースを使用しており、問題を解決する方法がわかりません。複数のコンシューマーがバインドされた 1 つのキューを作成し、RabbitMQ で一度に 1 つのメッセージを 1 つのコンシューマーにのみ送信し、ACK を待ってから別のメッセージを他のコンシューマーに送信したいと考えています。

これによりスループットが低下し、本質的に他の消費者を飢えさせる可能性があることは理解していますが、私にとっては問題ありません。この奇妙な使用例の理由は、コンシューマーが対話するサービスが一度に 1 つの同時要求しか処理できないため、これを制限する方法が必要ですが、コンシューマーも予期せず終了する可能性があり、別のコンシューマーがメッセージの処理をピックアップする必要があるためです。これが発生した場合。オプションがあることは知っていprefetchますが、それでも複数のユーザーが排他的なキューを取得できますが、それらが私が望むものを達成できるかどうかはわかりません。これを行うようにRabbitMQを構成することは可能ですか?

RabbitMQ の初心者であり、Java の初心者です。

手動 ack を使用し、Java Spring AMQP 抽象化を使用して消費者キャンセル通知を処理するリスナーを作成しようとしています。Spring 抽象化を使用して両方のタスクを達成できますか?

キューからメッセージを取得してそのメッセージを処理するリスナーを作成したいと考えています (おそらくデータベースなどに書き込みます)。メッセージの処理が失敗したり、何らかの理由で完了できない場合に、拒否して再度キューに入れることができるように、手動の確認応答を使用することを計画しました。これまでのところ、Spring AMQP を使用して手動で ack/nack/reject するには、ChannelAwareMessageListener.

私は、RabbitMQ からのコンシューマー キャンセル通知を処理する必要があることを認識してChannelAwareMessageListenerいますが、これをコーディングする方法が実際にはわかりません。CCN を処理する唯一の方法は、メッセージ配信とキャンセルを処理できるchannel.basicConsume()新しいインスタンスを呼び出して渡すことにより、下位レベルの Java クライアント API を使用してコードを記述することです。DefaultConsumer

clientPropertiesまた、構成内の Bean からファクトリを取得しているため、ConnectionFactory(ブローカーに CCN を処理できることを伝えるために)を設定する方法もわかりません。

リスナーとコンテナの作成の私の擬似コードは以下のとおりです。

サンプルの指示に従っています。これは、 http : //static.springsource.org/spring-amqp/docs/1.2.x/reference/html/sample-apps.htmlStock Tradingで要求/応答メッセージングを使用する方法を概説しています。 #d4e742spring-amqp

たまたま返信メッセージを送信できるメッセージ駆動型 POJOであるクラスを見ると、ServerHandlerMessageProperties を注入できる場所がわかりません。content-type

このサンプルを微調整して、単純な文字列を装った xml メッセージを返したいのですがServerHandler、返信を RabbitMQ に投稿すると、次のようになります。

そして最終的に（私の推測では）これを読んでいるはずの私のクライアントは、オブジェクトがcontent_type正しく設定されていないためにオブジェクトを解析できずapplication/json、ユースケースが失敗したためnull、返信として返されます。

Q) spring-amqp Request/Reply Messaging で応答の content-type を設定する方法を知っている人はいますか?

アップデート#1

jsonMessageConverterQ) おそらく、requestQueue からメッセージを読み取るためにのみ使用され、responseQueue に返信を戻したい場合には使用されないようにする必要があるかもしれません。これは、現在の appContext.xml ファイルからの抜粋です。

一部のサービス間の通信に Spring-AMQP を使用しています。サービス内で、消費者を宣言しています。コンシューマーごとに 1 つのキューを使用して、対応するメッセージをディスパッチします。コンシューマ マネージャは、すべての既知のコンシューマを保持します。実際のメッセージ ディスパッチは、対応するメッセージ タイプをサブスクライブしているコンシューマーを検索するメッセージ ハンドラー上で実行されます。ほとんどすべてのユースケースで、サービスは 1 つのコンシューマーのみでメッセージをサブスクライブします。

特別なケースとして、メッセージの種類だけでなく、メッセージの送信元であるキューも区別する必要があります。

メッセージが読み取られたキュー名を何らかの形で取得することは可能ですか?

POJO を使用して spring amqp を介して基本的な要求/応答メッセージングをセットアップしました。これらは JSON メッセージコンバーターを介して RabbitMQ で交換されます。

これまでの応答 pojo のペイロードは単純な文字列でしたが、今では画像を返信する必要があります。

Q1) これを処理する最善の方法は何ですか? JSON メッセージのペイロード文字列として画像を base64 エンコードする必要がありますか?

Q2) 単純に画像を BytesMessage として返すことは可能ですか?

Q2.a) spring-amqp は、POJO を返すリスナーと BytesMessage を返すリスナーの 2 つの別個のリスナーを処理できますか? 私はそうは思わないでしょうが、ここに私が求めているものの周りの疑似コードがあります:

Q2.b) または ... MessageListener がオブジェクトを返す場合 (POJO の場合もあれば、BytesMessage の場合もある) ... spring-amqp フレームワークは、実行時にそれに応じて調整して、シリアル化された POJO を送り返すことができますか?場合によっては json として、場合によっては BytesMessageをそのままにしますか?

私のアプリケーションには、非常に大きくなる可能性のあるキューがあります。マシンの空き容量がなくなった場合はどうすればよいですか? 複数のマシンでキューを分割するにはどうすればよいですか? たぶん、RabbitMQ の哲学が異なるため、1 つの大きなキューではなく、複数のキューを作成する必要があります..?

ベスト、フラビオ

メッセージを送信し、Spring AMQPを使用してMessageListenerを使用してそれらのメッセージを消費するRabbitMQサーバーで実行されているサンプルSpring amqpプロデューサーを作成しました。ここでは、キューとメッセージの耐久性を false に設定します。アノテーションを使用して「durable」フラグを false に設定する方法を教えてください。

ここにサンプルコードがあります

前もって感謝します。

Spring AMQP 1.2 リリースを使用して、RabbitMQ ブローカーのパフォーマンス テストに取り組んでいます。誰でも同じ手順を説明でき、プロデューサーとコンシューマーの両方のクライアントとして Java を使用しています。