問題タブ [spring-amqp]

私は、RabbitMQとSpring Integrationを使用して、着信 JSON メッセージを処理しています。

構成の関連部分は次のようになります。

JSON コンバーターとして Jackson Databind を使用しています。

受信 JSON メッセージの構文が正しくない場合があります。これにより、次の (正しい) 例外が発生します。

例外は、errorHandler単にMessagePublishingErrorHandlertoである によって処理されerrorChannelます。

ここまでは順調ですね。問題は、メッセージをエラー ハンドラで処理しているにもかかわらず、メッセージがまだ AMQP クライアントによって拒否されていることです。その後、RabbitMQ はメッセージを何度も再配信します。配信不能キューを構成しても役に立ちませんでした。このシナリオを正しく処理する方法はありますか?

(逆シリアル化が成功した後の) 処理のさらに下にある例外は問題なく処理されます。AMQP メッセージが確認され、エラー メッセージが に送信されerrorChannelます。

何か案は？

ライブラリのバージョン:

• 春の統合: 3.0.1
• 春のフレームワーク: 4.0.2
• ジャクソン データバインド: 2.3.1

名前付きキューからメッセージを受け取る必要があるメッセージ リスナーを作成しています。spring-amqp を使用して、バッキング交換の種類に関係なく、キューを介したメッセージの消費が発生するように、このリスナーを構成できるかどうか疑問に思っています。

例えば：

これにより、この例のキューがダイレクト エクスチェンジにバインドされますが、コンシューマーがキューをダイレクト、ファンアウト、ヘッダー、またはトピック エクスチェンジにバインドする必要がある場合はどうなるでしょうか? 基本的に、実行時にバッキング交換の正確なタイプを知る必要がないように、消費リスナーのキューバインディングを切り離す方法があるかどうか疑問に思っています。

私は初めて RabbitMQ を試しており、Spring AMQP ラッパーを使用することにしました。最初に書き込みパフォーマンスをテストし、送信者をキューに設定することを計画していました。

春のセットアップ:

送信者メイン:

ただし、15 個のメッセージを送信した後、クライアントは次の例外で失敗します。

どうして？

SimpleMessageListenerContainer複数のキューに接続され、 で設定された を使用しますChannelAwareMessageListener。メッセージが消費されたキューを特定することはできますか? 特に、メッセージが Exchange からキューにルーティングされた場合。

メッセージがキュー名を含むキューに直接送信されMessageProperties#getReceivedRoutingKeyた場合、メッセージが Exchange 経由でキューにルーティングされた場合、この情報には使用されたルーティング キーが含まれているようです。

メッセージがキューにどのように配信されたかに関係なく、この情報を正しく抽出できるメカニズムを探しています。または、RabbitMQ 側でこの情報を含むヘッダーを使用して情報を強化するメカニズム。

私は Spring AMQP を使用して、RabbitMQ でキューを作成しています。アプリが実行されているマシンの名前が名前に含まれるキューが必要です。したがって、アプリを実行する場所に応じて、キュー名は「fooQueue.host1」または「fooQueue.host2」になります。

私はこれを行う方法を考え出しました (以下で詳しく説明します) が、少し複雑に思えます。これを達成するためのより簡単/より良い/春の方法はありますか?

私の解決策

まず、マシン名を取得する Bean を作成します。

次に、Bean を Spring 構成に登録します。

次に、Spring AMQP 構成で次のように使用します。

RabbitMQ キューからのメッセージを消費する基本的な実装が与えられた場合、MessageListener処理中にスローされる可能性のある例外の種類に基づいて、メッセージを別の配信不能キューに送信するにはどうすればよいですか?

メッセージが最初に公開されたキューにはx-dead-letter-exchangeとx-dead-letter-routing-keyが設定されていますが、私の場合はこれだけでは不十分です。

念のために言うと、私のアプリケーションは Spring 4 と Spring Amqp を使用しています。

大きなバッチの処理を分散したい。アイデアは、Spring Batch を使用してクラウド内の一連の AMQP コンシューマーを起動し、安価なタスク (アイテム ID など) をロードして AMQP 交換に送信することです。結果の書き込みは、消費者自身によって行われます。

これを達成するための既製のライブラリはありますか?

いくつかの考え：

• Spring Batch は完全に交渉可能です。
• バッチサイズは数百万です。これらすべての ID を一度に総当たりで送信してメッセージ ブローカーを強制終了したくはありませんが、代わりになんらかのスロットリングを使用します。
• プロセスを監視できるように、どのアイテムが処理されたかを知りたいです。したがって、バッチ制御プロセスは、消費者からの応答を受信する必要があります。

春のウサギのmqテンプレートベースのリスナーを使用しています

負荷が増加すると、リスナーは単純にシャットダウンします

リスナーがシャットダウンしないようにするにはどうすればよいですか?

または、シャットダウン直後にリスナーを再起動する方法は?

私は次の設定を使用しています：

消費者は、次の関数を含む私のリスナーです