問題タブ [spring-amqp]

spring amqp を使用して rabbitmq を使用しようとしています。以下は私の構成です。

これは単純な Message Listener クラスです。

こちらはプロデューサー(春バッチのitemWriter)、

春のバッチ ジョブを実行すると、ImportItemWriter.write が呼び出されます。ただし、 ImportMessageListener.onMessage は機能しません。メッセージを出力しません。コンソールのすべての項目について以下の出力が得られます

HDFS にログ イベントをロードする単純な春の XD アプリケーションを試しています。spring-ampq/rabbit log4j appender(クラス) を使用してターゲット アプリケーションを構成org.springframework.amqp.rabbit.log4j.AmqpAppenderし、事前に構成された交換にログ メッセージを送り込みました。これらのメッセージを HDFS からプルし、HDFS にプッシュするように次のストリームを設定します。ここでは、ソース モジュールとシンク モジュールの両方が既製の XD モジュールです。

ストリーム定義、

新しいストリーム「demoQ1」を作成してデプロイしました

AMQP アペンダーは、メッセージを exchange にパブリッシュし、demoQ1 キューにルーティングします。ここで、rabbit ソースが最初のメッセージを取得し、メッセージを確認しないためスタックします。その理由は何ですか？

spring amqp を使用して rabbitmq を使用しようとしています。以下は私の構成です。

これは単純な Message Listener クラスです。

こちらはプロデューサー(春バッチのitemWriter)、

春のバッチ ジョブを実行すると、各メッセージが 1 つずつ送信されて処理され、応答を下回ります。

5 つのメッセージを送信してキューに入れ、5 つのコンシューマー スレッド (同時実行数 = 5) がそれらを同時に処理し、完了するとすぐに応答する必要があります。

プロデューサーが最初のメッセージの返信を待って2番目のメッセージをキューに入れたくありません。

非同期にする convertAndSend を使用してみましたが (返信を待機しません)、 convertSendAndReceive で取得できるように、itemWriter で返信メッセージを取得するにはどうすればよいですか?

テンプレート構成を次のように変更すると、

そして、 template.convertAndSend(item.toString()); を使用する場合 では、どうすれば返信メッセージを取得できますか?

このリスナーに独自のメッセージ ハンドラーをアタッチして、コンシューマー側でアタッチできる方法で応答メッセージを取得することはできません。返信には、デフォルトの RabbitmqTemplate ハンドラが必要です。

私は春のamqp rabbitmqを使用しており、メッセージを送信しています

私のメッセージハンドラは

私はそれを使用してフックしました

しかし、私は実行を下回っています

同様の問題がプロデューサー側にもあります。

また、 importExchange に例外がある場合、 replyHandler で例外を取得するにはどうすればよいですか?

すべてのメッセージを rabbitmq キューに入れ、リモート サーバーで処理しています。以下は、同じクラスのプロデューサーと応答ハンドラーです。

すべてのメッセージを write メソッドで送信し、onMessage で返信を取得しています。これは正常に機能していますが、書き込みは応答を待たず、呼び出し元に戻り、スプリング バッチ ステップが完了としてマークされます。

しかし、onMessage ですべての返信が得られるまで、すべてのメッセージを送信した後、プロセスが返信を待つようにしたいと考えています。どうすればこれを行うことができますか?

春のバッチ アイテム ライターで春の amqp テンプレートを使用して、rabbitmq キューにメッセージを追加しています。

メッセージは別のリモート サーバーで処理されます。メッセージ処理が失敗した場合 (何らかの例外が原因で)、ステップの実行が停止されるユースケースを処理しようとしています。

キュー内の残りのメッセージも失敗するため、キュー内の残りのメッセージを消費して処理しないように、そのキュー内の残りのメッセージをすべてパージしたいと考えています。

ステップが失敗した場合、アイテム ライターは再びすべてのメッセージをキューに入れるため、例外が発生した場合は残りのメッセージをすべて消去する必要があります。

spring amqp を使用してキューをパージするにはどうすればよいですか?