問題タブ [spring-amqp]

AMQP 0-9-1 プロトコルで 0-9-1 の背後にある正確なロジックを見つけることができませんでした。誰か説明してください。

Java クライアントとして Spring amqp 1.1 バージョンを使用しています。約 2000 のメッセージを持つキューがあります。このキューのサイズをチェックし、空の場合は「すべてのアイテムが処理されました」というメッセージを送信するサービスが必要です。

現在のキュー サイズを取得する方法がわかりません。助けてください

以前のバージョン 1.0 に存在していたクラス「RabbitBrokerAdmin」をググって見つけました。現在、1.1には存在しないと思います。

現在のキュー サイズを取得するための指針はありますか?

RabbitMQ にメッセージを送信する Java アプリケーションを作成しました。次に、Flume は RabbitMQ キューからメッセージを取得します。Flume を除いて、誰もキューからメッセージをプルしないことに興味があります。

私のアプリケーションは Spring AMQP Java プラグインを使用しています。

問題：

以下のコードでは、メッセージが RabbitMQ キューに送られ、「不明」のままになります。私が理解しているように、RabbitMQ は MessageListener からの ACK を待っていますが、MessageListener は決して ACK しません。誰かがそれを修正する方法を知っていますか?

コード：

開発環境を Linux から Mac に移行したところ、突然、RabbitMQ の奇妙な動作に直面しました。

RabbitMQ 2.8.7-1 が実行されている Ubuntu ボックスがあり、同じボックスで開発作業を行いました。私はテストコード (以下に提供) を実行しており、Rabbit はすべてのメッセージを喜んで受信しました。

ここで、以前の Linux ボックスと同じ構成で RabbitMQ を実行する Linux (ホスト B) で VirtualBox を実行している Mac ボックス (ホスト A) に移動しました。VM内でRabbitを呼び出すMacで開発環境を実行しています:)。しかし、何も来なかったので、Wireshark を使用して通信を追跡しました。

メッセージが受信されたようですが、ブローカーによって処理されていない可能性がありますか? クライアント側にログオンすると、メッセージが公開されたことがわかります。

問題の処理を開始する場所がまったくわかりません。以前は機能していたのに、現在は機能していないのはなぜですか。どこに問題がある可能性がありますか?

編集: OK、RabbitMQ チュートリアルから単純な送信者を実装しようとしましたが、アプリケーションがまだ実行されており、close() の後のコードに到達していないため、close() でハングしているようです。

Spring Integration を使用して以下を構成しようとしています。

1. チャネルにメッセージを送信します。
2. このメッセージを、n 個のコンシューマーとのラビット ファンアウト (pub/sub) 交換にパブリッシュします。
3. 各コンシューマーは応答メッセージを提供します。
4. 元のクライアントに返す前に、Spring Integration にこれらのレスポンスを集約させます。

これまでのところ、いくつかの問題があります...

1. 私はapply-sequence="true"、correlationId、sequenceSize、および sequenceNumber プロパティが設定されるようにプロパティを設定するために、publish-subscribe-channel を使用しています。これらのプロパティは、によって破棄されていDefaultAmqpHeaderMapperます。DEBUG headerName=[correlationId] WILL NOT be mapped

2. ファンアウト交換内に 2 つのキューが登録されているにもかかわらず、sequenceSize プロパティは 1 にしか設定されていません。おそらくこれは、メッセージがアグリゲータからリリースされるのが早すぎることを意味します。これは、使用するために publish-subscribe-channel を誤用しているためでapply-sequence="true"あり、サブスクライバーがint-amqp:outbound-gateway.

私の送信Spring構成は次のとおりです。

私の rabbitMQ 設定は次のとおりです。

コンシューマーは次のようになります。

どんな提案でも素晴らしいでしょう、私はスティックのどこかで間違った端を持っていると思います...

Gary のコメントに基づく新しいアウトバウンド スプリング構成:

1) サーバーがクライアントにメッセージを送信します。

2) インバウンドチャネルアダプターは、コンシューマーからの「MANUAL」承認モード操作を待機するように構成されています

3)「TaskBundlereceiver」ビーンは「ChannelAwareMessageListener」を実装しており、実装メソッドではメッセージ確認を行っています。

「TaskBundlereceiver」が実行されていません。何か不足していますか？

以下は、私が説明した手順の構成の詳細です。

あなたの意見に感謝します。

XML 構成:

spring-amqp を使用して、rabbitmq ノード (A、B、および C) のクラスターに接続します。クラスター内の rabbitmq ノードの 1 つがシャットダウンしたとします (C がシャットダウンします)。また、ノード C への接続を使用していた 2 つのメッセージ受信者 ( Receiver_1 と Receiver_2 ) があるとします。C がダウンすると、Receiver_1 と Receiver_2 は自動的に接続を切り替えて A または B に接続しますか? そして、C が再び現れるとどうなりますか? 一部のレシーバー接続は C を使用するように再配布されますか?

毎秒500以上の大量のストリームを処理しています。データはSpringAMQP+ Rabbitで、10個の同時コンシューマーを持つSimpleMessageListenerContainerを使用して消費されます。15分ごとにDbをチェックし、処理のために特定のプロパティをリロードする必要があります。これは、15分ごとに起動し、SimplelistenerContainerを停止し、必要な作業を実行して、コンテナーを再度開始するクォーツトリガーを使用して実行されます。

アプリの起動時にすべてが完全に機能し、トリガーが起動してコンテナーが再起動すると、同じメッセージが複数回配信されるのがわかります。これにより、多くの重複が発生します。消費者によって投げられる例外はありません。

メッセージリスナー

アプリの起動中に並列コンシューマーを設定し、コンシューマーを起動します

クォーツトリガー

と非同期でメッセージを消費していますSimpleMessageListenerContainer。そのメッセージはサービス層のメソッドを呼び出し、トランザクションのロールバックをテストするために意図的に例外をスローします。メッセージをキューの最後に送り返したいのですが、そうはなりません。誰かが私を助けることができますか？

Spring-AMQP - RabbitMQ アプリを開発しています。シャベルを使用してローカル キューからリモート キューにメッセージをレプリケートしたいのですが、Spring アプリからメッセージング サービスのほとんどすべての構成を行いたいと考えています。

この時点で、アプリ自体からキュー、交換などを作成できますが、同じアプリからシャベルを構成する方法が見つかりません。これを達成する方法はありますか？

シャベルの代わりにフェデレーション プラグインを使用する場合、Spring-AMQP を使用してシャベルを使用するよりも、それを構成するより良い方法はありますか?