問題タブ [spring-retry]

メッセージを失いたくないRabbitMQとSpring amqpを使用しています。再試行に指数関数的なバックオフ ポリシーを使用することで、コンシューマーが処理できるメッセージに取り組んでいる可能性があるコンシューマーをブロックする可能性があります。指数関数的なバックオフ ポリシーで失敗したメッセージを数日間再試行したいのですが、コンシューマーが数日間ブロックされることは望ましくなく、他のメッセージを処理し続けたいと考えています。

ActiveMQ(将来のある時点でメッセージを再試行する (ActiveMQ) )を使用してこの種の機能を実現できることはわかっていますが、RabbitMQ で同様のソリューションを見つけることができませんでした。

Spring amqp と RabbitMQ でこれを達成する方法はありますか?

リトライ メカニズムを処理するためのスプリング アスペクトを作成しました。Retry アノテーションも作成しました。以下は、Retry アノテーションのコードと、このアノテーションを処理するアスペクトです。

アノテーションを有効にするには、RetryInterceptor クラスをインスタンス化する必要があります。特定のコンテキストに対して、このオブジェクトのインスタンスが 1 つだけであることを確認したいと考えています。何らかの理由で複数のオブジェクトが作成された場合、私のアドバイスは何度も適用されます。常に 1 つのインスタンスが存在することを完全に保証するにはどうすればよいですか?

現在、正常に機能する Spring Integration フローがあります (図のリンクを参照)。現在の構成の上に Batch を追加して、指数関数的バックオフ、サーキット ブレーカー パターン、および再起動のためのデータベースへのジョブの永続化を使用して再試行できるようにしたいと考えています。

統合フローは、Message<MyObj> を受け取る Gateway で構成され、最終的に Message<MyObj> を Message<String> に変換する Transformer にルーティングされます。次に、アグリゲーターは Message<String> を受け取り、最終的に連結された Message<String> を解放します (サイズ解放戦略とタイムアウト付きの MessageGroupStoreReaper の両方を使用します)。連結された文字列は、SFTP outbound-channel-adapter を使用してアップロードされたファイルのペイロードです。

検索し、ドキュメントを読み、たくさんの例を見ましたが、プロセスの最後のステップをバッチ ジョブにカプセル化する方法がわかりません。アップロード中に SFTP 接続の問題やその他の例外がスローされた場合は、(ファイルのペイロードとして) 文字列のアップロードを再試行する機能が必要です。また、何らかの障害が発生した場合に (データベースでバックアップされた JobRepository を使用して) 再起動できるようにしたいので、再試行アドバイスを使用するだけでは不十分だと思います。

ピースをつなぎ合わせる方法と、どれを使用するか (ジョブ起動ゲートウェイ、MessageToJobRequest トランスフォーマー、ItemReader、ItemWriter??) を説明し、理解するのを手伝ってください。また、各 Message<String> にアクセスして、ジョブ、ステップ、またはタスクレット内の SFTP チャネルアダプターに送信する方法もわかりません。

現在の流れ:

現在、Spring AMQP バージョン1.3.6.RELEASEと Spring Retryを使用してい1.1.2.RELEASEます。Spring AMQP ドキュメント セクション3.3.1によると、を渡すことで再試行機能を追加できRetryTemplateます。

RecoveryCallback<T>実装を提供する既存の機能はありますか? RabbitTemplate.java の実装を確認していましたが、何も見つかりませんでした。

私が検討している使用例は*Send()、ブローカーがダウンしているために実行が失敗した場合に、独自のカスタム ロジックを実装したいというものです。

独自の実装でconvertAndSend()呼び出しをラップして実装できることは理解していますが、Spring AMQP がよりクリーンな実装を提供する場合、その道をたどりたくありませんでした。RetryTemplatetry { ... } catch (AmqpException e) { ... }

アプリケーションでRabbitMQを使用して支払いリクエストをキューに入れ、呼び出し元に結果を返す別のキューを用意しています。どちらの場合も、クライアントは永久に再試行する再試行ポリシーを要求しましたが、外部システムが監視によってバックアップを検出できるように、「x 回目のトランザクションを再試行しています...」のような内容を再試行のたびにログに記録します。ログファイル。

私はこのようにリスナーコンテナを作成しています:

再試行ロジックを次のように定義します。

したがって、再試行は問題なく機能しますが、実際に再試行で何かをログに記録するためのフックが見つかりません。足りないものはありますか？再試行時に何かを実行するためのフックがどこかにありますか? サブクラス化できるものはありますか？または、Spring の再試行ロジック内に埋もれている既存のログがあり、ロガー構成で有効にすることができますか?

ありがとう。

クリス。

私は春の再試行プロジェクトを試してきました。@Retryable 機能を正常に使用しましたが、JDK 動的プロキシを使用して機能させることができません。

テストで次のコード スニペットを使用しています。

ここで入手できるサンプル プロジェクト: https://github.com/maddenj-ie/retry.git

だから、私の質問は

1. これは、cglib または JDK 動的プロキシのいずれかを使用して機能する必要がありますか?

2. もしそうなら、私の設定の何が問題になっていますか?

助けてくれてありがとう。

よろしく、ジョー