問題タブ [aggregator]

いくつかの Java オブジェクトを分割してから集約しています。この補完戦略が Camel (2.15.2) でどのように機能するか、ちょっと混乱しています。完了サイズと完了タイムアウトを使用しています。私の理解が正しければ、完了タイムアウトはあまり影響しません。なぜなら、ここで待っていることはあまりないからです。

全体で、3000 以上のオブジェクトがあります。ただし、一部のみが集約されているようです。しかし、完成サイズの値を変えると状況が変わります。大きさが100なら800くらい、200なら1600くらいまで集まります。 ただ、オブジェクトの大きさは事前にわからないので、想定数に頼ることはできません。

ここで私が間違っていることを誰かに説明してもらえますか? 私がeagerCheckCompletionを使用すると、すべてが一度に集約されますが、これは望ましくありません。以下は私のルートです：

アグリゲーターの入力チャネルにメッセージを送信すると、アグリゲーターは集約されたメッセージを出力チャネルにリリースします。アグリゲーターは、(集約のために) 少なくとも 2 つのメッセージを予期しています。それ以外の場合は、タイムアウトになるまで 10 秒間待機します。jdbc メッセージ ストアも使用しています。

以下は、私がテストしたシナリオです。

シナリオ 1 は正常に動作しています

メッセージ 1 および 2 の送信 -> 入力チャネル (input1) -> アグリゲーター 1 -> 出力チャネル (output1)

シナリオ 2 は正常に動作しています

メッセージ 1 および 2 の送信 -> 入力チャネル (input2) -> アグリゲーター 2 -> 出力チャネル (output2)

シナリオ 3 は正常に動作しています

メッセージ 1 のみの送信 -> 入力チャネル (input1) -> アグリゲーター 1 -> 出力チャネル (output1)

期限切れのメッセージを output2 に送信する代わりに、output1 に送信しているため、シナリオ 4 は失敗します。

メッセージ 1 のみの送信 -> 入力チャネル (input2) -> アグリゲーター 2 -> 出力チャネル (output1)

シナリオ 4 が失敗する理由を誰かが提案できますか?

以下は私の構成です

前提 -

春の統合では、不完全なメッセージグループを持つアグリゲーターがある場合。グループ リリース戦略が満たされる前に、サーバーが再起動されます。

• 現在の動作 -> アグリゲーターに投稿されたすべてのメッセージは、新しいメッセージ グループではなく同じメッセージ グループに送られます。完了とマークされていないため、メッセージが流れ続けます。
• 予期される->サーバーが再起動された場合、アグリゲーターはメッセージストアから残りのメッセージを選択し、すでに保持されているメッセージを完了としてマークし、新しいメッセージに対応します。

私の期待は間違っていますか？誰かがガイドできますか？

パブリッシュ/サブスクライブ チャネルを使用して RESTful リクエストが 2 つのプロバイダーに転送される、非常に単純な統合フローがあります。両方の RESTful サービスからの結果は、1 つの配列に集約されます。統合フローのスケッチは次のとおりです。

ただし、私のコードを実行すると、結果の配列には RESTful サービスの 1 つだけから返された項目が含まれます。不足している構成手順はありますか?

アップデート

次のバージョンは、Artem のコメントを考慮して、完全なソリューションに対応しています。

特定のフィールドの整数値の範囲を生成するデータの混乱があるとしましょう...おそらくクラスタリングを行っているため、発生間隔のグループ化によってランク付けされたデータを確認したいと思います...次のように:

A) MongoDB アグリゲーターを使用して、B) 最小範囲と最大範囲にわたって等間隔でコレクション データをグループ化するための迅速で汚い方法をどのように作成できますか?

SQL クエリの出力を並行して処理しようとしています。以下は私のコードです。Aggregatorにsysoutがあります。しかし、Aggregator の sysout が印刷されていないことがランダムにわかります。また、アグリゲーターの解放方法も sysout を出力しません。どこかでメッセージを失っていると思います。誰でも光を当てることができますか？

JDBCMessageStore を使用するときの Aggregator の動作を理解しようとしています。以下は私のユースケースです：

1) キューからメッセージ (注文の詳細) を読み取ります。

1 つの注文メッセージには複数のorderItemが含まれます。

2) orderItemの数に基づいて注文メッセージを分割する

3) 4 つのorderItemメッセージに集約します。messageCountReleaseStrategy は、しきい値 = 4 で使用されます

4) 集約されたメッセージをサービス アクティベーターに転送します。

INT_MESSAGE_GROUP、INT_MESSAGE、および INT_GROUP_TO_MESSAGE の 3 つのテーブルのデータを観察していましたが、これらの 3 つのテーブルに最初にデータが挿入され、グループが解放されると、これらのテーブルから日付が削除/削除されるようです。

テーブルの 1 つ (INT_GROUP_TO_MESSAGE) で DB 挿入の失敗を確認したかったのです。エラーをシミュレートするために、テーブル INT_GROUP_TO_MESSAGE を削除しました。

以下のテストが実行されました。

1) 3 つのorderItemで 1 つの注文メッセージを投稿

2) メッセージでエラーが発生しました - テーブルが存在しません - (INT_GROUP_TO_MESSAGE が存在しないと予想されます)

3) 2 つのテーブルでレコードを見つけることができました

a) INT_MESSAGE_GROUP には 1 つのレコードがありました。

b) INT_MESSAGE には 23 のレコードがありました。メッセージが複数回再試行されたため、複数のエントリが行われたようです。

4) テーブル INT_GROUP_TO_MESSAGE を作成しました

5) 1 件のorderItemで 1 件の注文メッセージを投稿

6) 元のメッセージ (ステップ 1) と新しいメッセージ (ステップ 5) がキューから取得され、処理されました。

7) アグリゲーターは、グループに 4 つのメッセージをリリースしました。

8) ただし、データベース テーブル - INT_MESSAGEにはまだ 23 のレコードが存在しますが、他のテーブルは空です。

テーブルを削除するときに Db エラーをシミュレートしましたが、本番環境では、これらの挿入が失敗する原因となるメモリ、テーブルスペースの問題が発生する可能性があります。

私の質問は、INT_MESSAGE_GROUP、INT_MESSAGE、および INT_GROUP_TO_MESSAGE の 3 つのテーブルにトランザクションがあるように、任意のパラメーターを設定できますか?

以下は私の構成です

1 つの大きな入力ファイルが多くの個別のファイルにバーストされ、Spring Integration Aggregator を使用して再び集約される Spring Integration プロジェクトがあります。

パイプラインには、不要な個々のファイルを除外する多数のフィルターがあります。相関入力ファイルごとにフィルタリングされたファイルの数を追跡します。@ReleaseStrategy は、個々のファイルの数からフィルター処理された個々のファイルの数を引いた数を受け取ったかどうかを確認します。

ReleaseStrategy に到達する前に、処理される最後の個々のファイルがフィルター処理されるとどうなりますか? ReleaseStrategy は、それに到達する個々のファイルごとに呼び出されます。最後の個々のファイルがフィルター処理された場合、再度ポーリングされることはありませんが、Spring がこのユース ケースを予測し、まだ提供している非ハック的な規定を作成していることも期待できます。 @Aggregator イベントで私。タイムアウトした場合、またはすべてのフィルター ポイントが最後のファイルであるかどうかを確認するように設定した場合、@Aggregator イベントが発生しません。

ありがとう！