問題タブ [aggregator]

私は Apache Camel を使用するのが初めてです。ファイルの移動を手伝ってくれませんか？私はそのようなルートを持っています:

したがって、エラーのあるファイルは「失敗」ディレクトリに移動し、エラーのないファイルは「成功」ディレクトリに移動する必要があります。必要なメッセージを集約した後 (エラーのあるファイルの解析中に) 例外を生成しようとしたため、ファイルをディレクトリに移動するのに「失敗」しましたが、例外があったとしても、すべてのファイルがディレクトリに「成功」​​しました。

アグリゲーターの前に例外をスローすると、ファイルは「失敗した」ディレクトリに移動しましたが、最後の「to」(メールの送信) は機能しません。

この質問は SI フォーラムのこの投稿に関連していますが、フォーラムが閉鎖されているため、スレッドを続けるためにここに投稿します。

http://forum.spring.io/forum/spring-projects/integration/748192-messages-not-flowing-when-using-jms-channels

要約すると、Redis メッセージ ストアを備えたアグリゲーターと、60 秒ごとにスケジュールされたリーパーがあります。メッセージは、JMS チャネルを使用してアグリゲーターに送信されます。構成は次のとおりです。

私はスレッド ダンプに少し慣れていませんが、DefaultMessageListener スレッドが MessageReaper を起動する taskScheduler によってスタックされていることがわかります。特に、ReentrantLock クラスで。

何か案が？たぶん、これを避けるために他の構成を作成する必要があります

どんな助けでも大歓迎です

前もって感謝します！グスマン

ドキュメントの重複を排除するためのアグリゲーターがあります：

このクエリの順序はドキュメント スコアです。ただし、_timestamp フィールドで並べ替えたい。それが可能だ ？Date Histogram アグリゲーターでテストしました。しかし、成功せずに。

次のワークフローがあります。

1. インバウンドチャネル

2. スプリッター

3. 分割チャネルのタスク エグゼキュータ - すべてのスレッドが同じワークフローを実行します。

3.a. リクエストを構築する

3.b. ゲートウェイ メッセージ エンドポイントのサービス アクティベーター ラッパー。

3.c. error-channel の構成を使用した http-outbound-gateway のゲートウェイ ラッパー (http-outbound-gateway の呼び出し中に例外を処理するため)

3.d. http-アウトバウンド-ゲートウェイ

4. アグリゲータ

5. 春の統合ワークフローからの応答。

3.d で例外が発生した場合、ゲートウェイ エラー チャネル用に構成されたサービス アクティベーターに制御が移ります。失敗したメッセージから新しいヘッダー、エラー チャネルに渡されたヘッダーに次のコードだけをコピーします。

a. ｂ．シーケンス番号 ｃ．シーケンスサイズ

ただし、スプリッターの応答を集約している間、DefaultAggregatingMessageGroupProcessor.java は競合するヘッダーを削除し、それにより、アグリゲーターに制御を提供する前にエラー チャネルと応答チャネルを削除します。

したがって、アグリゲーターが操作を完了すると、応答またはエラー チャネルを見つけることができず、例外が発生します。

spring-integration-core バージョン 2.2.1 を使用していますが、ヘッダー アグリゲーション中に応答チャネルとエラー チャネルが削除される理由がわかりません。

この問題を解決するための情報は非常に役立ちます。

ありがとうございました ：）

編集 1: このシナリオで私を助けてくれてありがとう、ゲイリー。現在の構成を共有しています

私が気付いたのは、ゲートウェイを通過するときにすべての分割チャネルがエラーと応答用の新しい一時チャネルを作成し、ゲートウェイから応答を取得した後、保存された (元の受信) エラーと応答チャネルヘッダーを保持することです。そして、あなたが言及したように、制御がエラートランスフォーマーに到達した後、保存されたヘッダーを保持するフローが壊れ、集約メッセージグループプロセッサが一時チャネルの 3 つの異なるインスタンスを受信するため、それらを削除します。カスタム メッセージ グループ プロセッサを用意し、ヘッダーを集約するための競合解決戦略を変更することを計画していたので、この構成を思いつきました。

私はまだこれをテストしていません。しかし、この議論に基づくと、これは実行可能な解決策のようには見えません。

また、ゲートウェイでのエラー処理の構成が正しくないようです。ただし、「メッセージを直接転送する代わりに、エラーフローでエラーを処理し、結果を通常どおりゲートウェイの「ラッパー」に返す」というあなたの声明には混乱しています。エラー チャネルを削除すると、例外が発生したときにどのように制御を取り戻すことができますか? ここで何かを理解していない可能性があります。これについて詳しく教えてください。

私のビジネス ロジックは、アグリゲーターで構成されたグループ タイムアウトを変更できる必要があります。

コードは次のようになります。

問題は：

• 現在の値をユーザーに表示したいのですが...
• ゲッターは保護されています。

考えられるソリューション シナリオ:

• この Bean に式を注入してから、その値を変更する必要があります。
  • つまり、次にメッセージ グループを作成するときに新しい結果で式を再評価しますか?
• パブリック セッターを持つ独自のハンドラーで AggregatingMessageHandler を拡張する必要がありますか?
• または、これはバグであり、次のリリースで修正する必要がありますか?

私自身の実験に基づいて (これはどこにも文書化されていません)、同じ相関 ID とシーケンス番号を持つ 2 つのメッセージの場合、アグリゲーターは最初のメッセージのみを取得し、他のメッセージを破棄/無視します。

代わりに最後に受信したメッセージをアグリゲータに使用させる方法はありますか?

集約により、ペイロードが 1 つの文字列にマージされます。

単純なシナリオ: 相関 ID が同じでシーケンス サイズが 2 の 3 つのメッセージを受信時間順に並べたもの

• シーケンス番号: 1; ペイロード: abc
• シーケンス番号: 1; ペイロード: デフ
• シーケンス番号: 2; ペイロード: ghi

現在の出力: abcghi

期待される出力: defghi

このシナリオは、最初のメッセージ用のシーケンス番号 2 が欠落している場合に発生します。また、相関 ID (デコードされたペイロードから取得) は非常に限られているため、複数回使用されます。

オリジナルメッセージ

生のメッセージは次の形式で送信されます。

• 「配列サイズ」、「配列番号」、「ID」、「テキスト」
• ID の範囲は 0 ～ 9 です
• メッセージの例: 2,1,8,abc

生メッセージのペイロードの例:

• 2,1,8,abc
• 2,1,8,定義
• 2,2,8,ギ

アグリゲーターは基本的にテキストを結合します