# mulesoft Aggregators Module - Full Documentation # Aggregators Module 1.0 互換性と解決された問題に関する情報は、「Aggregator Module リリースノート」を参照してください。 Aggregators Module には、指定された設定を使用して値の保存と解放の両方を行うさまざまな種別のアグリゲーターが含まれています。​_アグリゲータースコープ_​は、指定された条件が満たされるまでこれらの値を収集するコンポーネントであり、条件が満たされると、集約された要素のリストをペイロードとして使用してプロセッサーのコンポーネントチェーンを実行します。 基本的なアグリゲーター設定は、集約する (式の評価を使用してメッセージから抽出される) 値の定義、集約がいつ完了したと見なされるかを定義する条件、およびその条件が満たされると実行されるプロセッサーのチェーンで構成されます。 アグリゲーターはパススルールーターです。アグリゲータースコープで受信したものと同じデータが、アグリゲータースコープの後のコンポーネントで処理されます。唯一の変更として、集約ルートの実行中に変数が設定された場合の変数の伝搬があります。 ## [](#始める前に)始める前に このモジュールを使用するには、以下に精通している必要があります。 - Anypoint Connector - Mule Runtime Engine (Mule) - Mule フローの要素とグローバル要素 - Anypoint Studio (Studio) を使用した Mule アプリケーションの作成 アプリケーションを作成するには、Anypoint Platform および Anypoint Studio が必要です。 ## [](#関連情報)関連情報 - [Anypoint Connector の概要](https://docs.mulesoft.com/connectors/introduction/introduction-to-anypoint-connectors) - [Exchange を使用したコネクタ、テンプレート、およびサンプルの検出](https://docs.mulesoft.com/connectors/introduction/intro-use-exchange) - [MuleSoft ヘルプセンター](https://help.mulesoft.com) # Aggregators Module 1.1 互換性と解決された問題に関する情報は、「Aggregator Module リリースノート」を参照してください。 Aggregators Module には、指定された設定を使用して値の保存と解放の両方を行うさまざまな種別のアグリゲーターが含まれています。​_アグリゲータースコープ_​は、指定された条件が満たされるまでこれらの値を収集するコンポーネントであり、条件が満たされると、集約された要素のリストをペイロードとして使用してプロセッサーのコンポーネントチェーンを実行します。 基本的なアグリゲーター設定は、集約する (式の評価を使用してメッセージから抽出される) 値の定義、集約がいつ完了したと見なされるかを定義する条件、およびその条件が満たされると実行されるプロセッサーのチェーンで構成されます。 アグリゲーターはパススルールーターです。アグリゲータースコープで受信したものと同じデータが、アグリゲータースコープの後のコンポーネントで処理されます。唯一の変更として、集約ルートの実行中に変数が設定された場合の変数の伝搬があります。 ## [](#始める前に)始める前に このモジュールを使用するには、以下に精通している必要があります。 - Anypoint Connector - Mule Runtime Engine (Mule) - Mule フローの要素とグローバル要素 - Anypoint Studio (Studio) を使用した Mule アプリケーションの作成 アプリケーションを作成するには、Anypoint Platform および Anypoint Studio が必要です。 ## [](#関連情報)関連情報 - [Anypoint Connector の概要](https://docs.mulesoft.com/connectors/introduction/introduction-to-anypoint-connectors) - [Exchange を使用したコネクタ、テンプレート、およびサンプルの検出](https://docs.mulesoft.com/connectors/introduction/intro-use-exchange) - [MuleSoft ヘルプセンター](https://help.mulesoft.com) # Aggregators Module の XML および Maven サポート XML で Mule アプリケーションを手動でコーディングすることもできますが、Anypoint Studio を使用した方が効率的です。 Anypoint Studio の XML エディターまたはテキストエディターから Mule Runtime Engine (Mule) アプリケーションを手動でコーディングする場合、参照ステートメントを XML Mule フローと Apache Maven ​`pom.xml`​ ファイルの両方に追加することで、アプリケーションでコネクタにアクセスできます。 コネクタを Studio に追加すると、Studio は自動的に XML コードにコネクタの名前空間およびスキーマの場所を入力します。さらに、​`pom.xml`​ ファイルに連動関係を追加します。 ## [](#コネクタの名前空間の追加)コネクタの名前空間の追加 次のコードを設定 XML のヘッダーの ​``​ タグ内に貼り付けます。 ```xml http://www.mulesoft.org/schema/mule/ http://www.mulesoft.org/schema/mule//current/mule-.xsd ``` 次の例は、​``​ XML ブロック内に名前空間ステートメントが配置される方法を示しています。 ```xml ``` ## [](#pom-ファイルの連動関係の追加)POM ファイルの連動関係の追加 Apache Maven ​`pom.xml`​ ファイルは Anypoint Studio によって生成され、Mule アプリケーションの連動関係が含まれます。Mule アプリケーションを手動でコーディングする場合、次の XML スニペットを ​`pom.xml`​ ファイルに含めてこのコネクタにアクセスできるようにします: ```xml org.mule.modules mule-aggregators-module x.x.x mule-plugin ``` `x.x.x`​ を使用しているコネクタに対応するバージョンに置き換えます。 最新の ​`pom.xml`​ ファイル情報を取得する手順は、次のとおりです。 1. [Anypoint Exchange](https://www.mulesoft.com/exchange/)​ に移動します。 2. Exchange で、​**\[Login (ログイン)\]**​ をクリックし、Anypoint Platform のユーザー名とパスワードを指定します。 3. Exchange で、「​`aggregators`​」を検索します。 4. コネクタを選択します。 5. 画面の右上付近にある ​**\[Dependency Snippets (連動関係スニペット)\]**​ をクリックします。 ## [](#次のステップ)次のステップ 名前空間および ​`pom.xml`​ ファイルを完了したら、​[例](https://docs.mulesoft.com/jp/aggregators-module/1.0/aggregators-examples)​を試すことができます。 ## [](#関連情報)関連情報 - [Anypoint Connector の概要](https://docs.mulesoft.com/connectors/introduction/introduction-to-anypoint-connectors) - [MuleSoft ヘルプセンター](https://help.mulesoft.com) # Aggregators Module の XML および Maven サポート XML で Mule アプリケーションを手動でコーディングすることもできますが、Anypoint Studio を使用した方が効率的です。 Anypoint Studio の XML エディターまたはテキストエディターから Mule Runtime Engine (Mule) アプリケーションを手動でコーディングする場合、参照ステートメントを XML Mule フローと Apache Maven ​`pom.xml`​ ファイルの両方に追加することで、アプリケーションでコネクタにアクセスできます。 コネクタを Studio に追加すると、Studio は自動的に XML コードにコネクタの名前空間およびスキーマの場所を入力します。さらに、​`pom.xml`​ ファイルに連動関係を追加します。 ## [](#コネクタの名前空間の追加)コネクタの名前空間の追加 次のコードを設定 XML のヘッダーの ​``​ タグ内に貼り付けます。 ```xml http://www.mulesoft.org/schema/mule/ http://www.mulesoft.org/schema/mule//current/mule-.xsd ``` 次の例は、​``​ XML ブロック内に名前空間ステートメントが配置される方法を示しています。 ```xml ``` ## [](#pom-ファイルの連動関係の追加)POM ファイルの連動関係の追加 Apache Maven ​`pom.xml`​ ファイルは Anypoint Studio によって生成され、Mule アプリケーションの連動関係が含まれます。Mule アプリケーションを手動でコーディングする場合、次の XML スニペットを ​`pom.xml`​ ファイルに含めてこのコネクタにアクセスできるようにします: ```xml org.mule.modules mule-aggregators-module x.x.x mule-plugin ``` `x.x.x`​ を使用しているコネクタに対応するバージョンに置き換えます。 最新の ​`pom.xml`​ ファイル情報を取得する手順は、次のとおりです。 1. [Anypoint Exchange](https://www.mulesoft.com/exchange/)​ に移動します。 2. Exchange で、​**\[Login (ログイン)\]**​ をクリックし、Anypoint Platform のユーザー名とパスワードを指定します。 3. Exchange で、「​`aggregators`​」を検索します。 4. コネクタを選択します。 5. 画面の右上付近にある ​**\[Dependency Snippets (連動関係スニペット)\]**​ をクリックします。 ## [](#次のステップ)次のステップ 名前空間および ​`pom.xml`​ ファイルを完了したら、​[例](https://docs.mulesoft.com/jp/aggregators-module/latest/aggregators-examples)​を試すことができます。 ## [](#関連情報)関連情報 - [Anypoint Connector の概要](https://docs.mulesoft.com/connectors/introduction/introduction-to-anypoint-connectors) - [MuleSoft ヘルプセンター](https://help.mulesoft.com) # Aggregators Module のトラブルシューティング Aggregators Module のトラブルシューティングを行うには、アプリケーションログに関する情報を理解し、冗長ログを有効にし、一般的なスローメッセージを解釈します。 ## [](#最新の-aggregators-module-バージョンの使用)最新の Aggregators Module バージョンの使用 確実に最新のパッチを使用するには、Aggregator Module を最新バージョンに更新します。Anypoint Studio によって生成される Apache Maven ​`pom.xml`​ ファイルでは、最新バージョンに対応するバージョンで ​`x.x.x`​ が置き換えられます。 ```xml org.mule.modules mule-aggregators-module x.x.x mule-plugin ``` 最新の ​`pom.xml`​ ファイル情報を取得する手順は、次のとおりです。 1. [Anypoint Exchange](https://www.mulesoft.com/exchange/)​ に移動します。 2. Exchange で、​**\[Login (ログイン)\]**​ をクリックし、Anypoint Platform のユーザー名とパスワードを指定します。 3. Exchange で、「​`aggregators`​」を検索します。 4. コネクタを選択します。 5. 画面の右上付近にある ​**\[Dependency Snippets (連動関係スニペット)\]**​ をクリックします。 ## [](#冗長ログの有効化)冗長ログの有効化 Aggregator Module のトラブルシューティングを開始するには、冗長ログを有効にして正確なエラーメッセージを表示します。 1. Anypoint Studio にアクセスし、​**\[Package Explorer\]**​ ビューに移動します。 2. アプリケーションのプロジェクトを開きます。 3. `src/main/resources`​ パスフォルダーを開きます。 4. フォルダー内の ​`log4j2.xml`​ ファイルを開きます。 5. 次の行がすでに ​`log4j2.xml`​ ファイル内にある場合は、そのコメントを解除して有効にします。それ以外の場合、次の行を追加します。 ```xml ``` 6. 変更を保存します。 7. **Package Explorer**​ でプロジェクト名をクリックし、​**\[Run (実行)\]**​ > ​**\[Run As (別のユーザーとして実行)\]**​ > ​**\[Mule Application (Mule アプリケーション)\]**​ をクリックします。 8. **\[Console (コンソール)\]**​ ビューに移動し、ロガーのメッセージを参照します。 ## [](#サイズベースのアグリゲーターでのタイムアウト動作の理解)サイズベースのアグリゲーターでのタイムアウト動作の理解 **\[Size based aggregator (サイズベースのアグリゲーター)\]**​ スコープでは、​**\[Timeout (タイムアウト)\]**​ 項目は集約が完了するのを待機する最大時間を示します。要素の合計数が ​**\[Max size (最大サイズ)\]**​ 項目で設定された値に達する前にタイムアウトすると、集約は完了とみなされず、リセットされます。​**\[Aggregator listener (アグリゲーターリスナー)\]**​ ソースを使用して、タイムアウトの発生前に集約された要素を取得することができます。 タイムアウト動作についての詳細は、​[「サイズベースのアグリゲーターのタイムアウト項目と最大サイズ項目の設定」](https://docs.mulesoft.com/jp/aggregators-module/1.0/aggregators-size-timeout)​を参照してください。 ## [](#複数のワーカーが設定されていないことを確認)複数のワーカーが設定されていないことを確認 CloudHub ではクラスタリングを使用できないため、アグリゲーターは複数のワーカーおよび永続的なオブジェクトストアでは機能しません。アプリケーションを拡大し、高可用性を提供するためにワーカーを共有または二重化する方法についての詳細は、​[「CloudHub Fabric」](https://docs.mulesoft.com/cloudhub/cloudhub-fabric)​ドキュメントを参照してください。 ## [](#一般的なスローの理解)一般的なスローの理解 ここでは、一般的なスローメッセージとその解決方法を示します。 **サイズベースのアグリゲーター**​の場合: - AGGREGATORS:AGGREGATOR\_CONFIG \[Max size (最大サイズ)\] または \[Timeout (タイムアウト)\] 項目の値が無効です (最大サイズが 0 以下など)。 - AGGREGATORS:OBJECT\_STORE\_ACCESS 集約された値の保存に使用されるオブジェクトストアへのアクセス中にエラーが発生しました。 **時間ベースのアグリゲーター**​の場合: - AGGREGATORS:AGGREGATOR\_CONFIG \[Period (期間)\] または \[Max size (最大サイズ)\] 項目の値が無効です (期間が 0 など)。 - AGGREGATORS:OBJECT\_STORE\_ACCESS 集約された値の保存に使用されるオブジェクトストアへのアクセス中にエラーが発生しました。 **グループベースのアグリゲーター**​の場合: - AGGREGATORS:GROUP\_COMPLETED まだ除去されていない完了済みのグループへの新しい要素の追加中にエラーが発生しました。 - AGGREGATORS:GROUP\_TIMED\_OUT タイムアウトしてまだ除去されていないグループへの新しい要素の追加中にエラーが発生しました。 - AGGREGATORS:NO\_GROUP\_ID グループ ID に解決される式が null を返しました。 - AGGREGATORS:NO\_GROUP\_SIZE グループサイズに解決される式が null を返しました。 - AGGREGATORS:AGGREGATOR\_CONFIG \[Group size (グループサイズ)\] または \[Timeout (タイムアウト)\] 項目の値が無効です (グループサイズが 0 以下など)。 - AGGREGATORS:OBJECT\_STORE\_ACCESS 集約された値の保存に使用されるオブジェクトストアへのアクセス中にエラーが発生しました。 ## [](#関連情報)関連情報 - [MuleSoft ヘルプセンター](https://help.mulesoft.com) - [Aggregators Module リファレンス](https://docs.mulesoft.com/jp/aggregators-module/1.0/aggregators-module-reference) # Aggregators Module のトラブルシューティング Aggregators Module のトラブルシューティングを行うには、アプリケーションログに関する情報を理解し、冗長ログを有効にし、一般的なスローメッセージを解釈します。 ## [](#最新の-aggregators-module-バージョンの使用)最新の Aggregators Module バージョンの使用 確実に最新のパッチを使用するには、Aggregator Module を最新バージョンに更新します。Anypoint Studio によって生成される Apache Maven ​`pom.xml`​ ファイルでは、最新バージョンに対応するバージョンで ​`x.x.x`​ が置き換えられます。 ```xml org.mule.modules mule-aggregators-module x.x.x mule-plugin ``` 最新の ​`pom.xml`​ ファイル情報を取得する手順は、次のとおりです。 1. [Anypoint Exchange](https://www.mulesoft.com/exchange/)​ に移動します。 2. Exchange で、​**\[Login (ログイン)\]**​ をクリックし、Anypoint Platform のユーザー名とパスワードを指定します。 3. Exchange で、「​`aggregators`​」を検索します。 4. コネクタを選択します。 5. 画面の右上付近にある ​**\[Dependency Snippets (連動関係スニペット)\]**​ をクリックします。 ## [](#冗長ログの有効化)冗長ログの有効化 Aggregator Module のトラブルシューティングを開始するには、冗長ログを有効にして正確なエラーメッセージを表示します。 1. Anypoint Studio にアクセスし、​**\[Package Explorer\]**​ ビューに移動します。 2. アプリケーションのプロジェクトを開きます。 3. `src/main/resources`​ パスフォルダーを開きます。 4. フォルダー内の ​`log4j2.xml`​ ファイルを開きます。 5. 次の行がすでに ​`log4j2.xml`​ ファイル内にある場合は、そのコメントを解除して有効にします。それ以外の場合、次の行を追加します。 ```xml ``` 6. 変更を保存します。 7. **Package Explorer**​ でプロジェクト名をクリックし、​**\[Run (実行)\]**​ > ​**\[Run As (別のユーザーとして実行)\]**​ > ​**\[Mule Application (Mule アプリケーション)\]**​ をクリックします。 8. **\[Console (コンソール)\]**​ ビューに移動し、ロガーのメッセージを参照します。 ## [](#サイズベースのアグリゲーターでのタイムアウト動作の理解)サイズベースのアグリゲーターでのタイムアウト動作の理解 **\[Size based aggregator (サイズベースのアグリゲーター)\]**​ スコープでは、​**\[Timeout (タイムアウト)\]**​ 項目は集約が完了するのを待機する最大時間を示します。要素の合計数が ​**\[Max size (最大サイズ)\]**​ 項目で設定された値に達する前にタイムアウトすると、集約は完了とみなされず、リセットされます。​**\[Aggregator listener (アグリゲーターリスナー)\]**​ ソースを使用して、タイムアウトの発生前に集約された要素を取得することができます。 タイムアウト動作についての詳細は、​[「サイズベースのアグリゲーターのタイムアウト項目と最大サイズ項目の設定」](https://docs.mulesoft.com/jp/aggregators-module/latest/aggregators-size-timeout)​を参照してください。 ## [](#複数のワーカーが設定されていないことを確認)複数のワーカーが設定されていないことを確認 CloudHub ではクラスタリングを使用できないため、アグリゲーターは複数のワーカーおよび永続的なオブジェクトストアでは機能しません。アプリケーションを拡大し、高可用性を提供するためにワーカーを共有または二重化する方法についての詳細は、​[「CloudHub Fabric」](https://docs.mulesoft.com/cloudhub/cloudhub-fabric)​ドキュメントを参照してください。 ## [](#一般的なスローの理解)一般的なスローの理解 ここでは、一般的なスローメッセージとその解決方法を示します。 **サイズベースのアグリゲーター**​の場合: - AGGREGATORS:AGGREGATOR\_CONFIG \[Max size (最大サイズ)\] または \[Timeout (タイムアウト)\] 項目の値が無効です (最大サイズが 0 以下など)。 - AGGREGATORS:OBJECT\_STORE\_ACCESS 集約された値の保存に使用されるオブジェクトストアへのアクセス中にエラーが発生しました。 **時間ベースのアグリゲーター**​の場合: - AGGREGATORS:AGGREGATOR\_CONFIG \[Period (期間)\] または \[Max size (最大サイズ)\] 項目の値が無効です (期間が 0 など)。 - AGGREGATORS:OBJECT\_STORE\_ACCESS 集約された値の保存に使用されるオブジェクトストアへのアクセス中にエラーが発生しました。 **グループベースのアグリゲーター**​の場合: - AGGREGATORS:GROUP\_COMPLETED まだ除去されていない完了済みのグループへの新しい要素の追加中にエラーが発生しました。 - AGGREGATORS:GROUP\_TIMED\_OUT タイムアウトしてまだ除去されていないグループへの新しい要素の追加中にエラーが発生しました。 - AGGREGATORS:NO\_GROUP\_ID グループ ID に解決される式が null を返しました。 - AGGREGATORS:NO\_GROUP\_SIZE グループサイズに解決される式が null を返しました。 - AGGREGATORS:AGGREGATOR\_CONFIG \[Group size (グループサイズ)\] または \[Timeout (タイムアウト)\] 項目の値が無効です (グループサイズが 0 以下など)。 - AGGREGATORS:OBJECT\_STORE\_ACCESS 集約された値の保存に使用されるオブジェクトストアへのアクセス中にエラーが発生しました。 ## [](#関連情報)関連情報 - [Aggregators Module リファレンス](https://docs.mulesoft.com/jp/aggregators-module/latest/aggregators-module-reference) - [MuleSoft ヘルプセンター](https://help.mulesoft.com) - [Known Issues](https://help.salesforce.com/s/issues) # Aggregators Module の例 Aggregators Module を最大限活用するには、モジュールのスコープ、ソース、および項目の設定に関する次の例を参照してください。 - [サイズベースのアグリゲータースコープの設定](https://docs.mulesoft.com/jp/aggregators-module/1.0/aggregators-size-example) 定義済みのサイズ数の要素の集約が完了するまで要素を集約するには、​**\[Size based aggregator (サイズベースのアグリゲーター)\]**​ スコープを設定します。 - [時間ベースのアグリゲータースコープの設定](https://docs.mulesoft.com/jp/aggregators-module/1.0/aggregators-time-example) 定義済みの時間制限内で要素を集約するには、​**\[Time based aggregator (時間ベースのアグリゲーター)\]**​ スコープを設定します。 - [グループベースのアグリゲータースコープの設定](https://docs.mulesoft.com/jp/aggregators-module/1.0/aggregators-group-example) 要素をグループ ID でグループに集約するように ​**\[Group based aggregator (グループベースのアグリゲーター)\]**​ スコープを設定します。 - [サイズベースのアグリゲーターのタイムアウト項目と最大サイズ項目の設定](https://docs.mulesoft.com/jp/aggregators-module/1.0/aggregators-size-timeout) **\[Size based aggregator (サイズベースのアグリゲーター)\]**​ スコープの ​**\[Timeout (タイムアウト)\]**​ 項目と ​**\[Max size (最大サイズ)\]**​ 項目を設定して、アグリゲーターのタイムアウト動作を理解します。 - [Aggregators Module の複数の例の設定](https://docs.mulesoft.com/jp/aggregators-module/1.0/aggregators-multiple-examples) 要素を集約するように ​**\[Size based aggregator (サイズベースのアグリゲーター)\]**​ スコープと ​**\[Group based aggregator (グループベースのアグリゲーター)\]**​ スコープを設定してバッチで処理します。 ## [](#関連情報)関連情報 - [Anypoint Connector の概要](https://docs.mulesoft.com/connectors/introduction/introduction-to-anypoint-connectors) - [MuleSoft ヘルプセンター](https://help.mulesoft.com) # Aggregators Module の例 Aggregators Module を最大限活用するには、モジュールのスコープ、ソース、および項目の設定に関する次の例を参照してください。 - [サイズベースのアグリゲータースコープの設定](https://docs.mulesoft.com/jp/aggregators-module/latest/aggregators-size-example) 定義済みのサイズ数の要素の集約が完了するまで要素を集約するには、​**\[Size based aggregator (サイズベースのアグリゲーター)\]**​ スコープを設定します。 - [時間ベースのアグリゲータースコープの設定](https://docs.mulesoft.com/jp/aggregators-module/latest/aggregators-time-example) 定義済みの時間制限内で要素を集約するには、​**\[Time based aggregator (時間ベースのアグリゲーター)\]**​ スコープを設定します。 - [グループベースのアグリゲータースコープの設定](https://docs.mulesoft.com/jp/aggregators-module/latest/aggregators-group-example) 要素をグループ ID でグループに集約するように ​**\[Group based aggregator (グループベースのアグリゲーター)\]**​ スコープを設定します。 - [サイズベースのアグリゲーターのタイムアウト項目と最大サイズ項目の設定](https://docs.mulesoft.com/jp/aggregators-module/latest/aggregators-size-timeout) **\[Size based aggregator (サイズベースのアグリゲーター)\]**​ スコープの ​**\[Timeout (タイムアウト)\]**​ 項目と ​**\[Max size (最大サイズ)\]**​ 項目を設定して、アグリゲーターのタイムアウト動作を理解します。 - [Aggregators Module の複数の例の設定](https://docs.mulesoft.com/jp/aggregators-module/latest/aggregators-multiple-examples) 要素を集約するように ​**\[Size based aggregator (サイズベースのアグリゲーター)\]**​ スコープと ​**\[Group based aggregator (グループベースのアグリゲーター)\]**​ スコープを設定してバッチで処理します。 ## [](#関連情報)関連情報 - [Anypoint Connector の概要](https://docs.mulesoft.com/connectors/introduction/introduction-to-anypoint-connectors) - [MuleSoft ヘルプセンター](https://help.mulesoft.com) # Aggregators Module の複数の例 アグリゲーターの一般的なユースケースは、複数の値を保存して後でバッチとして処理することです。 この集約を実現するには、​**\[Size based aggregator (サイズベースのアグリゲーター)\]**​ スコープを使用できます。 集約するデータは、以前に ​`personName`​ という変数に保存された人物名であり、100 要素のバッチを処理するとします。すべてのデータが収集されたら、​`whatToDoWithAllTheseNames`​ というフロー内で定義されたいくつかのロジックを実行します。 この場合、次を使用できます。 ```xml #[vars.personName] ``` 実行数が多すぎてしまうため、すべての新しい要素でアクションを実行しないようにするには、​`incremental-aggregation`​ を使用します。たとえば、次に示すように名前のリストが 10% ずつ増加するたびにアクションを実行するとします。 ```xml #[vars.personName] ``` 実行されるロジックは ​`partialAction`​ という別のフロー内にあるとします。 すべてのバッチが埋まるまで永久に待つことを避けるには、アグリゲーターのタイムアウトを設定して、​`maxSize`​ に達していなくても特定の時点ですべてのイベントを処理します。次の例では、タイムアウトを 10 秒に設定しています。 ```xml #[vars.personName] ``` タイムアウトが発生したときに、すべての集約された要素を処理するロジックが実行されるように、次のようなリスナーも含める必要があります。 ```xml ``` この場合、​`aggregation-complete`​ ルートと ​`aggregator-listener`​ から取得されるデータ型は同じです。そのため、フロー ​`whatToDoWithAllTheseNames`​ の実行では、データの取得元は関係ありません。 ただしこの設定では、集約が完了した場合、フロー ​`whatToDoWithAllTheseNames`​ は ​`aggregation-complete`​ ルート ​`flow-ref`​ から 1 回、​`aggregator-listener`​ でのフロー内の ​`flow-ref`​ からもう 1 回の計 2 回トリガーされます。 同じロジックが複数回実行されないようにするには、不完全な集約を確認するため、​`isAggregationComplete`​ 属性を使用する ​`choice`​ コンポーネントで ​`flow-ref`​ 値 ​`aggregator-listener`​ をラップする必要があります: ```xml ... #[vars.personName] ... ``` この例では、​`flow-ref`​ はタイムアウトでそのフローがトリガーされたときにのみコールされます。 最後に、国籍に従って各人物を処理する必要があります。この情報は変数で渡されます。ただし、この場合は同じ国のすべての人物を同じバッチでまとめて処理するとします。 グループベースのアグリゲーターでそれらを適宜分割できます。たとえば、人物の国籍が以前に ​`personNac`​ という変数に保存されたとします。この場合、フローは次のようになります。 ```xml #[vars.personName] ``` このフローにはタイムアウト用のリスナーも必要です。 ```xml ``` この例では、アグリゲーターで除去時間を設定していません。これは、デフォルト値が望ましい動作の ​`0`​ (グループの解放直後に除去) であるためです。そのため、グループが完了またはタイムアウトした場合、後続の値は拒否されることなくグループに追加されます。 ## [](#see_also)関連情報 - [Aggregators Module の例](https://docs.mulesoft.com/jp/aggregators-module/1.0/aggregators-examples) - [Aggregators Module リファレンス](https://docs.mulesoft.com/jp/aggregators-module/1.0/aggregators-module-reference) # Aggregators Module の複数の例 アグリゲーターの一般的なユースケースは、複数の値を保存して後でバッチとして処理することです。 この集約を実現するには、​**\[Size based aggregator (サイズベースのアグリゲーター)\]**​ スコープを使用できます。 集約するデータは、以前に ​`personName`​ という変数に保存された人物名であり、100 要素のバッチを処理するとします。すべてのデータが収集されたら、​`whatToDoWithAllTheseNames`​ というフロー内で定義されたいくつかのロジックを実行します。 この場合、次を使用できます。 ```xml #[vars.personName] ``` 実行数が多すぎてしまうため、すべての新しい要素でアクションを実行しないようにするには、​`incremental-aggregation`​ を使用します。たとえば、次に示すように名前のリストが 10% ずつ増加するたびにアクションを実行するとします。 ```xml #[vars.personName] ``` 実行されるロジックは ​`partialAction`​ という別のフロー内にあるとします。 すべてのバッチが埋まるまで永久に待つことを避けるには、アグリゲーターのタイムアウトを設定して、​`maxSize`​ に達していなくても特定の時点ですべてのイベントを処理します。次の例では、タイムアウトを 10 秒に設定しています。 ```xml #[vars.personName] ``` タイムアウトが発生したときに、すべての集約された要素を処理するロジックが実行されるように、次のようなリスナーも含める必要があります。 ```xml ``` この場合、​`aggregation-complete`​ ルートと ​`aggregator-listener`​ から取得されるデータ型は同じです。そのため、フロー ​`whatToDoWithAllTheseNames`​ の実行では、データの取得元は関係ありません。 ただしこの設定では、集約が完了した場合、フロー ​`whatToDoWithAllTheseNames`​ は ​`aggregation-complete`​ ルート ​`flow-ref`​ から 1 回、​`aggregator-listener`​ でのフロー内の ​`flow-ref`​ からもう 1 回の計 2 回トリガーされます。 同じロジックが複数回実行されないようにするには、不完全な集約を確認するため、​`isAggregationComplete`​ 属性を使用する ​`choice`​ コンポーネントで ​`flow-ref`​ 値 ​`aggregator-listener`​ をラップする必要があります: ```xml ... #[vars.personName] ... ``` この例では、​`flow-ref`​ はタイムアウトでそのフローがトリガーされたときにのみコールされます。 最後に、国籍に従って各人物を処理する必要があります。この情報は変数で渡されます。ただし、この場合は同じ国のすべての人物を同じバッチでまとめて処理するとします。 グループベースのアグリゲーターでそれらを適宜分割できます。たとえば、人物の国籍が以前に ​`personNac`​ という変数に保存されたとします。この場合、フローは次のようになります。 ```xml #[vars.personName] ``` このフローにはタイムアウト用のリスナーも必要です。 ```xml ``` この例では、アグリゲーターで除去時間を設定していません。これは、デフォルト値が望ましい動作の ​`0`​ (グループの解放直後に除去) であるためです。そのため、グループが完了またはタイムアウトした場合、後続の値は拒否されることなくグループに追加されます。 ## [](#see_also)関連情報 - [Aggregators Module の例](https://docs.mulesoft.com/jp/aggregators-module/latest/aggregators-examples) - [Aggregators Module リファレンス](https://docs.mulesoft.com/jp/aggregators-module/latest/aggregators-module-reference) # Aggregators Module リファレンス 1.0 リリースノート: [Aggregators Module リリースノート](https://docs.mulesoft.com/jp/release-notes/connector/aggregators-module-release-notes) _アグリゲータースコープ_​は、データを受信し、そのデータを処理して値を抽出し、その値を集約された要素のリストに追加するコンポーネントです。このプロセスが完了すると、アグリゲータースコープの設定に応じて、フロー内の一連のコンポーネントで要素のリストが処理されます。 アグリゲーターはパススルールーターです。アグリゲータースコープで受信したものと同じデータが、アグリゲータースコープの後のコンポーネントで処理されます。 唯一の変更として、集約ルートの実行中に変数が設定された場合の変数の伝搬があります。 アグリゲータースコープで保存値が解放されると、ルートコンポーネント (​**\[Incremental aggregation (増分集約)\]**​ または ​**\[Aggregation complete (集約完了)\]**​) で、​**\[Aggregator listener (アグリゲーターリスナー)\]**​ ソースを介してアグリゲータースコープ自体または別の Mule フロー内で、集約された要素のリストが処理されます。このプロセスは、使用されているアグリゲータースコープの設定によって異なります。 ## [](#設定)設定 Aggregator Module ではスコープが使用されるため、Aggregator Module のグローバル設定はありません。アグリゲーターの各種別は、そのパラメーターを介して設定可能な異なるスコープです。各スコープで共通に使用されるパラメーターを次に示します。 - Content (コンテンツ) 何を集約するかを定義する式。評価の結果は、集約に保存された値です。アグリゲータースコープに関連するすべてのデータはオブジェクトストアに保存されるため、コンテンツの値はシリアル化可能である必要があります。そうでない場合、モジュールが適切に動作しない場合があります。 - Object Store (オブジェクトストア) アグリゲーターに関連するすべての情報はオブジェクトストアに保存されます。オブジェクトストアの種別は、コンポーネントの想定される動作に関連します。 - 永続的なオブジェクトストア (デフォルト) メモリ内オブジェクトストアより遅くなりますが、信頼性は高くなります。アプリケーションの再起動後のデータ復旧が可能です。 - メモリ内オブジェクトストア 永続的なメモリストアよりも速くなりますが、アプリケーションが再起動するとすべてのデータが失われます。 ## [](#アグリゲータースコープ)アグリゲータースコープ サイズベース、時間ベース、およびグループベースのアグリゲーターがあります。 ### [](#size-based-aggregator)サイズベースのアグリゲーター **\[Size based aggregator (サイズベースのアグリゲーター)\]**​ スコープを使用すると、定義済みの数の要素の集約が完了するまで要素を集約することができます。 ```xml #[payload] ... ... ``` 要素が、指定された ​**\[Max size (最大サイズ)\]**​ に達すると、2 つのことが実行されます。 - ストレージ内の要素が削除され、次の要素は新しい集約に属します。 - 集約された要素を使用して ​**\[Aggregation complete (集約完了)\]**​ ルートが実行されます。 また、​**\[Aggregator listener (アグリゲーターリスナー)\]**​ がアグリゲータースコープに登録されている場合、同じ要素セットを使用してリスナーのコールバックが実行されます。 **\[Incremental aggregation (増分集約)\]**​ ルートが null ではなく、​**\[Max size (最大サイズ)\]**​ に達していない場合、集約された最後の要素を含めすべての集約された要素を使用してルートコンポーネントチェーンが実行されます。 アグリゲータースコープの ​**\[Timeout (タイムアウト)\]**​ を定義することもできます。この場合、このタイムアウトを遅延として使用するスケジュールされたタスクが実行用に登録されます。この時間は最初の要素が到着した時間から計算されます。また、実行を待機している別のタスクがある場合、追加のタスクはスケジュールされません。タイムアウトが発生すると、参照先のリスナーは、タイムアウトによって呼び出されることがサポートされている場合のみ、実行されます。 #### [](#パラメーター)パラメーター
名前 型 説明 デフォルト値 必須

Name (名前)

String (文字列)

アグリゲーターの名前。[Aggregator listener (アグリゲーターリスナー)] は後で名前でスコープを参照できます。

x

Content (コンテンツ)

式

何を集約するかを定義する式。評価の結果は、集約に保存された値です。

#[payload]

Max Size (最大サイズ)

Number (数値)

集約が完了とみなされる前に集約される要素の合計数。

x

Timeout (タイムアウト)

Number (数値)

集約が完了するのを待機する最大時間。要素の合計数が [Max Size (最大サイズ)] に達する前にタイムアウトすると、集約は完了とみなされません。グループが頻繁にタイムアウトすることを避けるため、​0​ の値はサポートされていません。

-1(UNLIMITED)

Timeout unit (タイムアウト単位)

Time Unit (時間単位)

タイムアウトを測定する時間単位。

SECONDS

Object Store (オブジェクトストア)

Object Store (オブジェクトストア)

集約された要素が保存されるグローバルオブジェクトストアを参照する名前または非公開オブジェクトストアの定義。

デフォルトのオブジェクトストアパーティション。

Aggregation Complete Route (集約完了ルート)

Route (ルート)

集約の完了時に実行するコンポーネントチェーン。タイムアウトした場合、集約完了ルートは実行されません。

x

Incremental Aggregation Route (増分集約ルート)

Route (ルート)

集約されたすべての新規要素で実行するコンポーネントチェーン。ペイロードは、この集約で集約されたすべての要素 (最初の要素から現在集約中の要素まで) のリストです。
#### [](#発生)発生 - AGGREGATORS:AGGREGATOR\_CONFIG \[Max size (最大サイズ)\] または \[Timeout (タイムアウト)\] 項目の値が無効です (​`maxSize < 0`​ など)。 - AGGREGATORS:OBJECT\_STORE\_ACCESS 集約された値の保存に使用されるオブジェクトストアへのアクセス中にエラーが発生しました。 ### [](#time-based-aggregator)時間ベースのアグリゲーター **\[Time based aggregator (時間ベースのアグリゲーター)\]**​ スコープを使用すると、指定した時間制限に達するまで要素を集約することができます。 ```xml #[payload] ... ``` 考慮される期間は、最初の要素が到着した時間から計算されます。集約が解放されても、次の要素が到着するまでタイマーは開始されません。 アグリゲーターでは、​**\[Max size (最大サイズ)\]**​ が設定されていない限り、新しい要素が到着するたびに ​**\[Incremental aggregation (増分集約)\]**​ ルートを実行することもできます。 この場合、​**\[Incremental aggregation (増分集約)\]**​ ルートは、集約された要素のサイズが ​**\[Max size (最大サイズ)\]**​ に達した場合を除き、毎回実行されます。この時点で ​**\[Aggregator listener (アグリゲーターリスナー)\]**​ が存在する場合、リスナーのコールバックも実行されます。 #### [](#パラメーター-2)パラメーター
名前 型 説明 デフォルト値 必須

Name (名前)

String (文字列)

アグリゲーターの名前。[Aggregator listener (アグリゲーターリスナー)] は後で名前でスコープを参照できます。

x

Content (コンテンツ)

式

何を集約するかを定義する式。評価の結果は、集約に保存された値です。

#[payload]

Period (期間)

Number (数値)

集約が完了とみなされるまで待機する期間。

x

Period unit (期間単位)

Time Unit (時間単位)

期間を測定する時間単位。

SECONDS

Max Size (最大サイズ)

Number (数値)

集約が完了とみなされる前に集約される要素の合計数。

-1(UNLIMITED)

Object Store (オブジェクトストア)

Object Store (オブジェクトストア)

集約された要素が保存されるグローバルオブジェクトストアを参照する名前または非公開オブジェクトストアの定義。

デフォルトのオブジェクトストアパーティション。

Incremental Aggregation Route (増分集約ルート)

Route (ルート)

集約されたすべての新規要素で実行するコンポーネントチェーン。ペイロードは、この集約で集約されたすべての要素 (最初の要素から現在集約中の要素まで) のリストです。
#### [](#発生-2)発生 - AGGREGATORS:AGGREGATOR\_CONFIG \[Period (期間)\] または \[Max size (最大サイズ)\] 項目の値が無効です (期間が 0 など)。 - AGGREGATORS:OBJECT\_STORE\_ACCESS 集約された値の保存に使用されるオブジェクトストアへのアクセス中にエラーが発生しました。 ### [](#group-based-aggregator)グループベースのアグリゲーター **\[Group based aggregator (グループベースのアグリゲーター)\]**​ スコープを使用すると、要素をグループ ID でグループに集約できます。 ```xml #[payload] ... ... ``` 要素が、グループに指定された ​**\[Max size (最大サイズ)\]**​ に達すると、2 つのことが実行されます。 - このグループの要素がストレージから削除されます。グループは完了としてマークされ、そのグループに到着するすべての新しい要素で例外が発生します。 - この特定のグループの集約された要素を使用して ​**\[Aggregation complete (集約完了)\]**​ ルートが実行されます。 新しい要素がアグリゲーターに到達するたびに、ID が解決されます。この ID のグループがすでにアグリゲーターに存在する場合、値はそのグループに追加されます。それ以外の場合、この ID の新しいグループが作成され、受信した要素がそのグループの集約の最初の要素になります。 また、​**\[Aggregator listener (アグリゲーターリスナー)\]**​ がアグリゲータースコープに登録されている場合、同じ要素セットを使用してリスナーのコールバックが実行されます。 **\[Incremental aggregation (増分集約)\]**​ ルートが null ではなく、​**\[Max size (最大サイズ)\]**​ に達していない場合、集約された最後の要素を含めすべての集約された要素を使用してルートコンポーネントチェーンが実行されます。 **\[Group based aggregator (グループベースのアグリゲーター)\]**​ スコープではいくつかの重要な概念が導入されます。 - グループのタイムアウト グループの必要なすべての要素が想定時間内に到着しなかったためにグループを解放する必要がある場合に指定します。グループがタイムアウトし、まだ除去されていない場合、新しい値を追加しようとする試みは拒否されます。 - グループの除去 完了したかタイムアウトしたかに関わらずグループをアグリゲーターから削除する場合に指定します。このグループの ID を持つ新しい要素をアグリゲーターが受信すると、そのグループが再度作成されます。 最後に、グループベースのアグリゲーターに到達する要素が、分割されたシーケンスから渡された場合 (​[ForEach](https://docs.mulesoft.com/mule-runtime/latest/for-each-scope-concept)​ コンポーネントによってなど)、各要素には異なる ​`sequenceNumber`​ が割り当てられます。この場合、要素は集約解放の前に昇順に並び替えられます。 #### [](#パラメーター-3)パラメーター
名前 型 説明 デフォルト値 必須

Name (名前)

String (文字列)

アグリゲーターの名前。[Aggregator listener (アグリゲーターリスナー)] は後で名前でスコープを参照できます。

x

Content (コンテンツ)

式

何を集約するかを定義する式。評価の結果は、集約に保存された値です。

#[payload]

Group Id (グループ ID)

式

メッセージを集約する必要があるグループの ID を取得するため、受信した各新規メッセージで評価される式。

#[correlationId]

Group Size (グループサイズ)

Number (数値)

解決されたグループ ID のグループに割り当てる最大サイズ。グループ ID が同じすべてのメッセージは、同じグループサイズにする必要があります。そうしないと、最初に解決されたグループサイズのみが正しいとみなされます。一致しないすべてのメッセージに対して警告が記録されます。

#[itemSequenceInfo.sequenceSize]

Eviction Time (除去時間)

Number (数値)

完了またはタイムアウトしたグループ ID を記憶する時間 (0: 記憶しない、-1: 永久に記憶)。

180

Eviction Time Unit (除去時間単位)

Time Unit (時間単位)

除去時間の時間単位。

SECONDS

Timeout (タイムアウト)

Number (数値)

グループの集約が完了するのを待機する最大時間。そのグループ内の要素の合計数がグループのサイズと同じになる前にタイムアウトすると、集約は完了とみなされます。グループが頻繁にタイムアウトすることを避けるため、​0​ の値はサポートされていません。

-1(UNLIMITED)

Timeout unit (タイムアウト単位)

Time Unit (時間単位)

タイムアウトを測定する時間単位。

SECONDS

Object Store (オブジェクトストア)

Object Store (オブジェクトストア)

集約された要素が保存されるグローバルオブジェクトストアを参照する名前または非公開オブジェクトストアの定義。

デフォルトのオブジェクトストアパーティション。

Aggregation Complete Route (集約完了ルート)

Route (ルート)

集約の完了時に実行するコンポーネントチェーン。

x

Incremental Aggregation Route (増分集約ルート)

Route (ルート)

集約されたすべての新規要素で実行するコンポーネントチェーン。ペイロードは、この集約で集約されたすべての要素 (最初の要素から現在集約中の要素まで) のリストです。
#### [](#発生-3)発生 - AGGREGATORS:GROUP\_COMPLETED まだ除去されていない完了済みのグループへの新しい要素の追加中にエラーが発生しました。 - AGGREGATORS:GROUP\_TIMED\_OUT タイムアウトしてまだ除去されていないグループへの新しい要素の追加中にエラーが発生しました。 - AGGREGATORS:NO\_GROUP\_ID グループ ID に解決される式が null を返しました。 - AGGREGATORS:NO\_GROUP\_SIZE グループサイズに解決される式が null を返しました。 - AGGREGATORS:AGGREGATOR\_CONFIG \[Group size (グループサイズ)\] または \[Timeout (タイムアウト)\] 項目の値が無効です (グループサイズが 0 以下など)。 - AGGREGATORS:OBJECT\_STORE\_ACCESS 集約された値の保存に使用されるオブジェクトストアへのアクセス中にエラーが発生しました。 ## [](#ソース)ソース ### [](#aggregator-listener)アグリゲーターリスナー **\[Aggregator listener (アグリゲーターリスナー)\]**​ は、アグリゲータースコープによりトリガーされる要素をリスンするソースです。 ``
[Aggregator listener (アグリゲーターリスナー)] は、フロー内にあるアグリゲータースコープのみを参照します。サブフローで宣言されたアグリゲータースコープを [Aggregator listener (アグリゲーターリスナー)] は参照できません。
リスナーで参照されるアグリゲータースコープが集約を完了すると、すべての要素のリストを使用してリスナーがトリガーされます。 **\[Aggregator listener (アグリゲーターリスナー)\]**​ はソースであるため、アグリゲーターとは異なるフローに配置されます。リスナーはアグリゲーターのフローのコンテキストにアクセスできないため、そのフローの変数にアクセスできません。 集約リスナーはあらゆるアグリゲーターで使用できますが、特に時間駆動型の非同期集約で重要です。非同期集約では、アグリゲータールートが実行されず、\[Aggregator listener (アグリゲーターリスナー)\] をソースとするフロー内のコンポーネントにのみアクセスできます。 #### [](#パラメーター-4)パラメーター
名前 型 説明 デフォルト値 必須

Aggregator Name (アグリゲーター名)

String (文字列)

リスンするアグリゲーターの名前。このアグリゲーターがその要素を解放すると、リスナーが実行されます。各リスナーが参照できるアグリゲーターは 1 つのみで、各アグリゲーターは最大 1 つのリスナーで参照可能です。

x

Include Timed Out Groups (タイムアウトグループを含める)

Boolean (ブール)

タイムアウトによりグループが解放されたときに、リスナーがトリガーされるかどうかを示します。

false
## [](#集約属性)集約属性 メッセージの集約が行われるたびに、その集約に関する情報を含む属性がメッセージにいくつか追加されます。
名前 型 説明

Aggregation ID (集約 ID)

String (文字列)

要素が集約されたグループの ID。集約戦略でグループ別に集約しない場合、この項目は集約が解放されるまで保持される自動生成値になります (グループベースおよび時間ベースのアグリゲーターと同様)。

First Item Arrival Time (最初の項目の到着時間)

Date (日付)

最初の値が集約された時間。

Last Item Arrival Time (最後の項目の到着時間)

Date (日付)

最後の値が集約された時間。

Is Aggregation Complete (集約完了)

Boolean (ブール)

集約が完了した場合は True、そうでない場合は False。
## [](#see_also)関連情報 - [Aggregators Module の例](https://docs.mulesoft.com/jp/aggregators-module/1.0/aggregators-examples) - [MuleSoft ヘルプセンター](https://help.mulesoft.com) # Aggregators Module リファレンス 1.1 リリースノート: [Aggregators Module リリースノート](https://docs.mulesoft.com/jp/release-notes/connector/aggregators-module-release-notes) _アグリゲータースコープ_​は、データを受信し、そのデータを処理して値を抽出し、その値を集約された要素のリストに追加するコンポーネントです。このプロセスが完了すると、アグリゲータースコープの設定に応じて、フロー内の一連のコンポーネントで要素のリストが処理されます。 アグリゲーターはパススルールーターです。アグリゲータースコープで受信したものと同じデータが、アグリゲータースコープの後のコンポーネントで処理されます。 唯一の変更として、集約ルートの実行中に変数が設定された場合の変数の伝搬があります。 アグリゲータースコープで保存値が解放されると、ルートコンポーネント (​**\[Incremental aggregation (増分集約)\]**​ または ​**\[Aggregation complete (集約完了)\]**​) で、​**\[Aggregator listener (アグリゲーターリスナー)\]**​ ソースを介してアグリゲータースコープ自体または別の Mule フロー内で、集約された要素のリストが処理されます。このプロセスは、使用されているアグリゲータースコープの設定によって異なります。 ## [](#設定)設定 Aggregator Module ではスコープが使用されるため、Aggregator Module のグローバル設定はありません。アグリゲーターの各種別は、そのパラメーターを介して設定可能な異なるスコープです。各スコープで共通に使用されるパラメーターを次に示します。 - Content (コンテンツ) 何を集約するかを定義する式。評価の結果は、集約に保存された値です。アグリゲータースコープに関連するすべてのデータはオブジェクトストアに保存されるため、コンテンツの値はシリアル化可能である必要があります。そうでない場合、モジュールが適切に動作しない場合があります。 - Object Store (オブジェクトストア) アグリゲーターに関連するすべての情報はオブジェクトストアに保存されます。オブジェクトストアの種別は、コンポーネントの想定される動作に関連します。 - 永続的なオブジェクトストア (デフォルト) メモリ内オブジェクトストアより遅くなりますが、信頼性は高くなります。アプリケーションの再起動後のデータ復旧が可能です。 - メモリ内オブジェクトストア 永続的なメモリストアよりも速くなりますが、アプリケーションが再起動するとすべてのデータが失われます。 ## [](#アグリゲータースコープ)アグリゲータースコープ サイズベース、時間ベース、およびグループベースのアグリゲーターがあります。 ### [](#size-based-aggregator)サイズベースのアグリゲーター **\[Size based aggregator (サイズベースのアグリゲーター)\]**​ スコープを使用すると、定義済みの数の要素の集約が完了するまで要素を集約することができます。 ```xml #[payload] ... ... ``` 要素が、指定された ​**\[Max size (最大サイズ)\]**​ に達すると、2 つのことが実行されます。 - ストレージ内の要素が削除され、次の要素は新しい集約に属します。 - 集約された要素を使用して ​**\[Aggregation complete (集約完了)\]**​ ルートが実行されます。 また、​**\[Aggregator listener (アグリゲーターリスナー)\]**​ がアグリゲータースコープに登録されている場合、同じ要素セットを使用してリスナーのコールバックが実行されます。 **\[Incremental aggregation (増分集約)\]**​ ルートが null ではなく、​**\[Max size (最大サイズ)\]**​ に達していない場合、集約された最後の要素を含めすべての集約された要素を使用してルートコンポーネントチェーンが実行されます。 アグリゲータースコープの ​**\[Timeout (タイムアウト)\]**​ を定義することもできます。この場合、このタイムアウトを遅延として使用するスケジュールされたタスクが実行用に登録されます。この時間は最初の要素が到着した時間から計算されます。また、実行を待機している別のタスクがある場合、追加のタスクはスケジュールされません。タイムアウトが発生すると、参照先のリスナーは、タイムアウトによって呼び出されることがサポートされている場合のみ、実行されます。 #### [](#パラメーター)パラメーター
名前 型 説明 デフォルト値 必須

Name (名前)

String (文字列)

アグリゲーターの名前。[Aggregator listener (アグリゲーターリスナー)] は後で名前でスコープを参照できます。

x

Content (コンテンツ)

Expression (式)

何を集約するかを定義する式。評価の結果は、集約に保存された値です。

#[payload]

Max Size (最大サイズ)

Number (数値)

集約が完了とみなされる前に集約される要素の合計数。

x

Timeout (タイムアウト)

Number (数値)

集約が完了するのを待機する最大時間。要素の合計数が [Max Size (最大サイズ)] に達する前にタイムアウトすると、集約は完了とみなされません。グループが頻繁にタイムアウトすることを避けるため、​0​ の値はサポートされていません。

-1(UNLIMITED)

Timeout unit (タイムアウト単位)

Time Unit (時間単位)

タイムアウトを測定する時間単位。

SECONDS

Object Store (オブジェクトストア)

Object Store (オブジェクトストア)

集約された要素が保存されるグローバルオブジェクトストアを参照する名前または非公開オブジェクトストアの定義。

デフォルトのオブジェクトストアパーティション。

Aggregation Complete Route (集約完了ルート)

Route (ルート)

集約の完了時に実行するコンポーネントチェーン。タイムアウトした場合、集約完了ルートは実行されません。

x

Incremental Aggregation Route (増分集約ルート)

Route (ルート)

集約されたすべての新規要素で実行するコンポーネントチェーン。ペイロードは、この集約で集約されたすべての要素 (最初の要素から現在集約中の要素まで) のリストです。
#### [](#発生)発生 - AGGREGATORS:AGGREGATOR\_CONFIG \[Max size (最大サイズ)\] または \[Timeout (タイムアウト)\] 項目の値が無効です (​`maxSize < 0`​ など)。 - AGGREGATORS:OBJECT\_STORE\_ACCESS 集約された値の保存に使用されるオブジェクトストアへのアクセス中にエラーが発生しました。 ### [](#time-based-aggregator)時間ベースのアグリゲーター **\[Time based aggregator (時間ベースのアグリゲーター)\]**​ スコープを使用すると、指定した時間制限に達するまで要素を集約することができます。 ```xml #[payload] ... ``` 考慮される期間は、最初の要素が到着した時間から計算されます。集約が解放されても、次の要素が到着するまでタイマーは開始されません。 アグリゲーターでは、​**\[Max size (最大サイズ)\]**​ が設定されていない限り、新しい要素が到着するたびに ​**\[Incremental aggregation (増分集約)\]**​ ルートを実行することもできます。 この場合、​**\[Incremental aggregation (増分集約)\]**​ ルートは、集約された要素のサイズが ​**\[Max size (最大サイズ)\]**​ に達した場合を除き、毎回実行されます。この時点で ​**\[Aggregator listener (アグリゲーターリスナー)\]**​ が存在する場合、リスナーのコールバックも実行されます。 #### [](#パラメーター-2)パラメーター
名前 型 説明 デフォルト値 必須

Name (名前)

String (文字列)

アグリゲーターの名前。[Aggregator listener (アグリゲーターリスナー)] は後で名前でスコープを参照できます。

x

Content (コンテンツ)

Expression (式)

何を集約するかを定義する式。評価の結果は、集約に保存された値です。

#[payload]

Period (期間)

Number (数値)

集約が完了とみなされるまで待機する期間。

x

Period unit (期間単位)

Time Unit (時間単位)

期間を測定する時間単位。

SECONDS

Max Size (最大サイズ)

Number (数値)

集約が完了とみなされる前に集約される要素の合計数。

-1(UNLIMITED)

Object Store (オブジェクトストア)

Object Store (オブジェクトストア)

集約された要素が保存されるグローバルオブジェクトストアを参照する名前または非公開オブジェクトストアの定義。

デフォルトのオブジェクトストアパーティション。

Incremental Aggregation Route (増分集約ルート)

Route (ルート)

集約されたすべての新規要素で実行するコンポーネントチェーン。ペイロードは、この集約で集約されたすべての要素 (最初の要素から現在集約中の要素まで) のリストです。
#### [](#発生-2)発生 - AGGREGATORS:AGGREGATOR\_CONFIG \[Period (期間)\] または \[Max size (最大サイズ)\] 項目の値が無効です (期間が 0 など)。 - AGGREGATORS:OBJECT\_STORE\_ACCESS 集約された値の保存に使用されるオブジェクトストアへのアクセス中にエラーが発生しました。 ### [](#group-based-aggregator)グループベースのアグリゲーター **\[Group based aggregator (グループベースのアグリゲーター)\]**​ スコープを使用すると、要素をグループ ID でグループに集約できます。 ```xml #[payload] ... ... ``` 要素が、グループに指定された ​**\[Max size (最大サイズ)\]**​ に達すると、2 つのことが実行されます。 - このグループの要素がストレージから削除されます。グループは完了としてマークされ、そのグループに到着するすべての新しい要素で例外が発生します。 - この特定のグループの集約された要素を使用して ​**\[Aggregation complete (集約完了)\]**​ ルートが実行されます。 新しい要素がアグリゲーターに到達するたびに、ID が解決されます。この ID のグループがすでにアグリゲーターに存在する場合、値はそのグループに追加されます。それ以外の場合、この ID の新しいグループが作成され、受信した要素がそのグループの集約の最初の要素になります。 また、​**\[Aggregator listener (アグリゲーターリスナー)\]**​ がアグリゲータースコープに登録されている場合、同じ要素セットを使用してリスナーのコールバックが実行されます。 **\[Incremental aggregation (増分集約)\]**​ ルートが null ではなく、​**\[Max size (最大サイズ)\]**​ に達していない場合、集約された最後の要素を含めすべての集約された要素を使用してルートコンポーネントチェーンが実行されます。 **\[Group based aggregator (グループベースのアグリゲーター)\]**​ スコープではいくつかの重要な概念が導入されます。 - グループのタイムアウト グループの必要なすべての要素が想定時間内に到着しなかったためにグループを解放する必要がある場合に指定します。グループがタイムアウトし、まだ除去されていない場合、新しい値を追加しようとする試みは拒否されます。 - グループの除去 完了したかタイムアウトしたかに関わらずグループをアグリゲーターから削除する場合に指定します。このグループの ID を持つ新しい要素をアグリゲーターが受信すると、そのグループが再度作成されます。 最後に、グループベースのアグリゲーターに到達する要素が、分割されたシーケンスから渡された場合 (​[ForEach](https://docs.mulesoft.com/mule-runtime/latest/for-each-scope-concept)​ コンポーネントによってなど)、各要素には異なる ​`sequenceNumber`​ が割り当てられます。この場合、要素は集約解放の前に昇順に並び替えられます。 #### [](#パラメーター-3)パラメーター
名前 型 説明 デフォルト値 必須

Name (名前)

String (文字列)

アグリゲーターの名前。[Aggregator listener (アグリゲーターリスナー)] は後で名前でスコープを参照できます。

x

Content (コンテンツ)

Expression (式)

何を集約するかを定義する式。評価の結果は、集約に保存された値です。

#[payload]

Group Id (グループ ID)

式

メッセージを集約する必要があるグループの ID を取得するため、受信した各新規メッセージで評価される式。

#[correlationId]

Group Size (グループサイズ)

Number (数値)

解決されたグループ ID のグループに割り当てる最大サイズ。グループ ID が同じすべてのメッセージは、同じグループサイズにする必要があります。そうしないと、最初に解決されたグループサイズのみが正しいとみなされます。一致しないすべてのメッセージに対して警告が記録されます。

#[itemSequenceInfo.sequenceSize]

Eviction Time (除去時間)

Number (数値)

完了またはタイムアウトしたグループ ID を記憶する時間 (0: 記憶しない、-1: 永久に記憶)。

180

Eviction Time Unit (除去時間単位)

Time Unit (時間単位)

除去時間の時間単位。

SECONDS

Timeout (タイムアウト)

Number (数値)

グループの集約が完了するのを待機する最大時間。グループ内の要素の合計数がグループのサイズと同じになる前にタイムアウトすると、アグリゲーターは ​notifyListenerOnTimeout​ をトリガーします。これにより、​IncludeTimedOutGroups​ が ​true​ に設定されている場合に、タイムアウトした集約を完了としてマークせずに、アグリゲーターリスナーによって取得できます。さらに、​IncludeTimedOutGroups​ パラメーターにより、集約を完了としてマークせずに、これらのタイムアウトグループをリスナーに渡すかどうかが決定されます。グループが頻繁にタイムアウトすることを避けるため、​0​ の値はサポートされていません。

-1(UNLIMITED)

Timeout unit (タイムアウト単位)

Time Unit (時間単位)

タイムアウトを測定する時間単位。

SECONDS

Object Store (オブジェクトストア)

Object Store (オブジェクトストア)

集約された要素が保存されるグローバルオブジェクトストアを参照する名前または非公開オブジェクトストアの定義。

デフォルトのオブジェクトストアパーティション。

Aggregation Complete Route (集約完了ルート)

Route (ルート)

集約の完了時に実行するコンポーネントチェーン。

x

Incremental Aggregation Route (増分集約ルート)

Route (ルート)

集約されたすべての新規要素で実行するコンポーネントチェーン。ペイロードは、この集約で集約されたすべての要素 (最初の要素から現在集約中の要素まで) のリストです。
#### [](#発生-3)発生 - AGGREGATORS:GROUP\_COMPLETED まだ除去されていない完了済みのグループへの新しい要素の追加中にエラーが発生しました。 - AGGREGATORS:GROUP\_TIMED\_OUT タイムアウトしてまだ除去されていないグループへの新しい要素の追加中にエラーが発生しました。 - AGGREGATORS:NO\_GROUP\_ID グループ ID に解決される式が null を返しました。 - AGGREGATORS:NO\_GROUP\_SIZE グループサイズに解決される式が null を返しました。 - AGGREGATORS:AGGREGATOR\_CONFIG \[Group size (グループサイズ)\] または \[Timeout (タイムアウト)\] 項目の値が無効です (グループサイズが 0 以下など)。 - AGGREGATORS:OBJECT\_STORE\_ACCESS 集約された値の保存に使用されるオブジェクトストアへのアクセス中にエラーが発生しました。 ## [](#ソース)ソース ### [](#aggregator-listener)アグリゲーターリスナー **\[Aggregator listener (アグリゲーターリスナー)\]**​ は、アグリゲータースコープによりトリガーされる要素をリスンするソースです。 ``
[Aggregator listener (アグリゲーターリスナー)] は、フロー内にあるアグリゲータースコープのみを参照します。サブフローで宣言されたアグリゲータースコープを [Aggregator listener (アグリゲーターリスナー)] は参照できません。
リスナーで参照されるアグリゲータースコープが集約を完了すると、すべての要素のリストを使用してリスナーがトリガーされます。 **\[Aggregator listener (アグリゲーターリスナー)\]**​ はソースであるため、アグリゲーターとは異なるフローに配置されます。リスナーはアグリゲーターのフローのコンテキストにアクセスできないため、そのフローの変数にアクセスできません。 集約リスナーはあらゆるアグリゲーターで使用できますが、特に時間駆動型の非同期集約で重要です。非同期集約では、アグリゲータールートが実行されず、\[Aggregator listener (アグリゲーターリスナー)\] をソースとするフロー内のコンポーネントにのみアクセスできます。 #### [](#パラメーター-4)パラメーター
名前 型 説明 デフォルト値 必須

Aggregator Name (アグリゲーター名)

String (文字列)

リスンするアグリゲーターの名前。このアグリゲーターがその要素を解放すると、リスナーが実行されます。各リスナーが参照できるアグリゲーターは 1 つのみで、各アグリゲーターは最大 1 つのリスナーで参照可能です。

x

Include Timed Out Groups (タイムアウトグループを含める)

Boolean (ブール)

タイムアウトによりグループが解放されたときに、リスナーがトリガーされるかどうかを示します。

false
## [](#集約属性)集約属性 メッセージの集約が行われるたびに、その集約に関する情報を含む属性がメッセージにいくつか追加されます。
名前 型 説明

Aggregation ID (集約 ID)

String (文字列)

要素が集約されたグループの ID。集約戦略でグループ別に集約しない場合、この項目は集約が解放されるまで保持される自動生成値になります (グループベースおよび時間ベースのアグリゲーターと同様)。

First Item Arrival Time (最初の項目の到着時間)

Date (日付)

最初の値が集約された時間。

Last Item Arrival Time (最後の項目の到着時間)

Date (日付)

最後の値が集約された時間。

Is Aggregation Complete (集約完了)

Boolean (ブール)

集約が完了した場合は True、そうでない場合は False。
### [](#RedeliveryPolicy)Redelivery Policy (再配信ポリシー) エラーを生成する要求を実行するための再配信ポリシーを設定します。再配信ポリシーは、フロー内のあらゆる提供元に追加できます。
項目 型 説明 デフォルト値 必須

Max Redelivery Count (最大再配信数)

Number (数値)

​REDELIVERY_EXHAUSTED​ エラーが返されるまでに、再配信された要求が処理に失敗できる最大回数。

Use Secure Hash (セキュアハッシュを使用)

Boolean (ブール)

​true​ の場合、再配信されたメッセージの識別にセキュアハッシュアルゴリズムが使用されます。

Message Digest Algorithm (メッセージダイジェストアルゴリズム)

String (文字列)

<b>[Use Secure Hash (セキュアハッシュを使用)]<b/> 項目が ​true​ の場合に使用するセキュアハッシュアルゴリズム。メッセージのペイロードが Java オブジェクトの場合、Mule ではこの値が無視され、ペイロードの ​hashCode()​ が返した値が返されます。

Id Expression (ID の式)

String (文字列)

メッセージがいつ再配信されたのかを判断する 1 つ以上の式。このプロパティは、<b>[Use Secure Hash (セキュアハッシュを使用)]</b> 項目が ​false​ の場合にのみ設定できます。

Object Store (オブジェクトストア)

Object Store (オブジェクトストア)

各メッセージの再配信カウンターが保存されるオブジェクトストアを設定します。
## [](#システムプロパティ)システムプロパティ
プロパティ 説明 デフォルト値 例

mule.aggregator.executor.failOnStartIfNoQuorum

現在、クォーラム制約が満たされていない場合、Mule アプリケーションのデプロイメントは失敗します。このシステムプロパティは、Mule アプリケーションをデプロイするためにバックグラウンドでクォーラムポーリングを送信します。

True

-M-Dmule.aggregator.executor.failOnStartIfNoQuorum=false

mule.aggregator.executor.delayForQuorum

クォーラムポーリング間の時間 (ミリ秒) を設定します。このプロパティは、​mule.aggregator.executor.failOnStartIfNoQuorum​ プロパティの値が ​false​ の場合に有効になります。

100

-M-Dmule.aggregator.executor.delayForQuorum=200

mule.aggregatorsSchedulingPeriod

サイズベースまたは時間ベースのアグリゲーターが設定されたサイズに達したかどうかを判断するためのスケジュール期間 (ミリ秒) を設定します。このモジュールは、1 秒ごとに GET コールをトリガーします。

-M-Dmule.aggregatorsSchedulingPeriod=1
## [](#see_also)関連情報 - [Aggregators Module の例](https://docs.mulesoft.com/jp/aggregators-module/latest/aggregators-examples) - [MuleSoft ヘルプセンター](https://help.mulesoft.com) # Anypoint Studio を使用した Aggregators Module 1.0 の設定 Anypoint Studio (Studio) エディターは、Mule アプリケーション、プロパティ、および設定ファイルの設計と更新に役立ちます。 Studio でモジュールを追加して設定する手順は、次のとおりです。 1. [Mule プロジェクトを作成します](#create-mule-project)​。 2. [モジュールを Mule プロジェクトに追加します](#add-connector-to-project)​。 3. [モジュールのフローのソースを設定します](#configure-source)​。 4. [アグリゲータースコープをフローに追加します](#add-connector-operation)​。 5. [アグリゲーターのオブジェクトストアを設定します](#configure-os-aggregator)​。 [「アプリケーションログの表示」](#view-app-log)​で説明されているように、コネクタを実行するときに、リアルタイムでアプリケーションログを表示して問題を確認できます。 Studio で初めてコネクタを設定する場合は、​[「Anypoint Studio を使用したコネクタの設定」](https://docs.mulesoft.com/connectors/introduction/intro-config-use-studio)​を参照してください。このトピックを参照した後、コネクタ項目についての詳細情報が必要な場合は、​[「Aggregators Module リファレンス」](https://docs.mulesoft.com/jp/aggregators-module/1.0/aggregators-module-reference)​を参照してください。 ## [](#create-mule-project)Mule プロジェクトを作成する Studio で、コネクタを追加および設定する新しい Mule プロジェクトを作成します。 1. Studio で、​**\[File (ファイル)\] > \[New (新規)\] > \[Mule Project (Mule プロジェクト)\]**​ を選択します。 2. Mule プロジェクトの名前を入力して、​**\[Finish (完了)\]**​ をクリックします。 ## [](#add-connector-to-project)モジュールを Mule プロジェクトに追加する Aggregators Module を Mule プロジェクトに追加して、XML コードにモジュールの名前空間およびスキーマの場所を自動的に入力し、プロジェクトの ​`pom.xml`​ ファイルに必須の連動関係を追加します。 1. **\[Mule Palette (Mule パレット)\]**​ ビューで、​**\[(X) Search in Exchange ((X) Exchange 内を検索)\]**​ をクリックします。 2. **\[Add Dependencies to Project (連動関係をプロジェクトに追加)\]**​ ウィンドウで、検索項目に「​`aggregators module`​」と入力します。 3. **\[Available modules (使用可能なモジュール)\]**​ で ​**\[Aggregators Module (Aggregator Module)\]**​ をクリックします。 4. **\[Add (追加)\]**​ をクリックします。 5. **\[Finish (完了)\]**​ をクリックします。 Studio でモジュールを Mule プロジェクトに追加しても、Studio ワークスペースの他のプロジェクトはそのモジュールを使用できません。 ## [](#configure-source)ソースの設定 ソースは、指定された条件が満たされたときにフローを開始します。 次のソースのいずれかを Aggregators Module で使用するように設定できます。 - **\[HTTP\] > \[Listener\]** 設定されたホストとポートで要求を受信するたびにフローを開始します。 - **Aggregator listener (アグリゲーターリスナー)** 参照されるアグリゲータースコープによりトリガーされる要素をリスンすることで、サブフローを開始します。 リスナーで参照されるアグリゲータースコープが集約を完了すると、​**\[Aggregator listener (アグリゲーターリスナー)\]**​ はすべての集約された要素のリストを使用してサブフローをトリガーします。 たとえば、\[Aggregator listener (アグリゲーターリスナー)\] を設定する手順は、次のとおりです。 1. **\[Mule Palette (Mule パレット)\]**​ ビューで、​**\[Aggregators (アグリゲーター)\] > \[Aggregator listener (アグリゲーターリスナー)\]**​ を選択します。 2. **\[Aggregator listener (アグリゲーターリスナー)\]**​ を Studio キャンバスにドラッグします。 3. **\[Aggregator listener (アグリゲーターリスナー)\]**​ 設定画面で、必要に応じて ​**\[Display Name (表示名)\]**​ 項目の値を変更します。 4. **\[Aggregator name (アグリゲーター名)\]**​ を、リスンするアグリゲータースコープの名前に設定します。 アグリゲータースコープがその要素を解放すると、リスナーが実行されます。 5. **\[Include timed out groups (タイムアウトグループを含める)\]**​ を選択し、タイムアウトによりグループが解放されたときに、リスナーがトリガーされるかどうかを示します。 この設定オプションは、​**\[Time based aggregator (時間ベースのアグリゲーター)\]**​ スコープや ​**\[Size based aggregator (サイズベースのアグリゲーター)\]**​ スコープなど、グループ以外のアグリゲーターでも機能します。 ## [](#add-connector-operation)アグリゲータースコープをフローに追加する フローにモジュールスコープを追加するときは、そのモジュールで実行するアクションを指定しています。 Aggregators Module のスコープを追加する手順は、次のとおりです。 1. **\[Mule Palette (Mule パレット)\]**​ ビューで ​**\[Aggregators (アグリゲーター)\]**​ を選択し、目的のスコープを選択します。 - **Group based aggregator (グループベースのアグリゲーター)** 要素をグループ ID でグループに集約できます。 - **Size based aggregator (サイズベースのアグリゲーター)** 定義済みのサイズ数の要素の集約が完了するまで要素を集約することができます。 - **Time based aggregator (時間ベースのアグリゲーター)** 定義済みの時間制限内で要素を集約できます。 2. スコープを Studio キャンバスのソースの右にドラッグします。 ![Anypoint Studio の \[Mule Palette (Mule パレット)\] ビュー内の Aggregator Module スコープとソース](_images/aggregators-scopes.png) Figure 1. Studio での Aggregator Module スコープ ## [](#configure-os-aggregator)アグリゲーターのオブジェクトストアを設定する どのアグリゲータースコープでも、グローバルオブジェクトストアを参照するか非公開オブジェクトストアを作成することで、オブジェクトストアを設定できます。 ### [](#オブジェクトストアを設定する)オブジェクトストアを設定する Studio でグローバルオブジェクトストアを設定するには、次の手順に従います。 1. フローからアグリゲータースコープを選択します。 2. アグリゲータースコープ設定画面で、​**\[Advanced (詳細)\]**​ をクリックします。 3. 緑のプラス記号 (+) をクリックして、グローバルオブジェクトストア設定を作成します。 4. **\[Global Object store configuration (グローバルオブジェクトストア設定)\]**​ 画面で、​**\[Name (名前)\]**​ を ​`aGlobalObjectStore`​ に設定します。 5. 必要な設定属性を入力して、​**\[OK\]**​ をクリックします。 次のスクリーンショットは、​**\[Group based aggregator (グループベースアグリゲーター)\]**​ スコープの ​**\[Advanced (詳細)\]**​ タブのオブジェクトストア設定を示しています。 ![「Object Store(\*) (オブジェクトストア(\*))」 項目では aGlobalObjectStore 設定が選択されています。](/mulesoft/current/ja-jp/aggregators-module/1.0/_images/aggregators-os-global.png) **\[Configuration XML (設定 XML)\]**​ ビューでは ​`objectStore`​ 設定は次のように表示されます。 ```xml ``` ### [](#非公開オブジェクトストアの設定)非公開オブジェクトストアの設定 Studio で非公開オブジェクトストアを設定するには、XML コードを手動で更新します。 1. フローからアグリゲータースコープを選択します。 2. Studio キャンバスで ​**\[Configuration XML (設定 XML)\]**​ をクリックします。 3. XML の非公開オブジェクトストア ​``​ を ​``​ セクションの内側に追加します。次に例を示します。 ```xml ... ``` ## [](#view-app-log)アプリケーションログの表示 問題を確認するため、アプリケーションログを次の方法で表示できます。 - アプリケーションを Anypoint Platform から実行している場合、アプリケーションログ出力は Anypoint Studio のコンソールウィンドウに表示されます。 - コマンドラインから Mule を使用してアプリケーションを実行している場合、アプリケーションログ出力は OS コンソールに表示されます。 アプリケーションのログファイル (​`log4j2.xml`​) でログファイルパスがカスタマイズされていない場合、デフォルトの場所 ​`MULE_HOME/logs/.log`​ でアプリケーションログを表示することもできます。 ## [](#関連情報)関連情報 - [Anypoint Connector の概要](https://docs.mulesoft.com/connectors/introduction/introduction-to-anypoint-connectors) - [Anypoint Studio を使用したコネクタの設定](https://docs.mulesoft.com/connectors/introduction/intro-config-use-studio) - [Aggregators Module リファレンス](https://docs.mulesoft.com/jp/aggregators-module/1.0/aggregators-module-reference) - [MuleSoft ヘルプセンター](https://help.mulesoft.com) # Anypoint Studio を使用した Aggregators Module 1.1 の設定 Anypoint Studio (Studio) エディターは、Mule アプリケーション、プロパティ、および設定ファイルの設計と更新に役立ちます。 Studio でモジュールを追加して設定する手順は、次のとおりです。 1. [Mule プロジェクトを作成します](#create-mule-project)​。 2. [モジュールを Mule プロジェクトに追加します](#add-connector-to-project)​。 3. [モジュールのフローのソースを設定します](#configure-source)​。 4. [アグリゲータースコープをフローに追加します](#add-connector-operation)​。 5. [アグリゲーターのオブジェクトストアを設定します](#configure-os-aggregator)​。 [「アプリケーションログの表示」](#view-app-log)​で説明されているように、コネクタを実行するときに、リアルタイムでアプリケーションログを表示して問題を確認できます。 Studio で初めてコネクタを設定する場合は、​[「Anypoint Studio を使用したコネクタの設定」](https://docs.mulesoft.com/connectors/introduction/intro-config-use-studio)​を参照してください。このトピックを参照した後、コネクタ項目についての詳細情報が必要な場合は、​[「Aggregators Module リファレンス」](https://docs.mulesoft.com/jp/aggregators-module/latest/aggregators-module-reference)​を参照してください。 ## [](#create-mule-project)Mule プロジェクトを作成する Studio で、コネクタを追加および設定する新しい Mule プロジェクトを作成します。 1. Studio で、​**\[File (ファイル)\] > \[New (新規)\] > \[Mule Project (Mule プロジェクト)\]**​ を選択します。 2. Mule プロジェクトの名前を入力して、​**\[Finish (完了)\]**​ をクリックします。 ## [](#add-connector-to-project)モジュールを Mule プロジェクトに追加する Aggregators Module を Mule プロジェクトに追加して、XML コードにモジュールの名前空間およびスキーマの場所を自動的に入力し、プロジェクトの ​`pom.xml`​ ファイルに必須の連動関係を追加します。 1. **\[Mule Palette (Mule パレット)\]**​ ビューで、​**\[(X) Search in Exchange ((X) Exchange 内を検索)\]**​ をクリックします。 2. **\[Add Dependencies to Project (連動関係をプロジェクトに追加)\]**​ ウィンドウで、検索項目に「​`aggregators module`​」と入力します。 3. **\[Available modules (使用可能なモジュール)\]**​ で ​**\[Aggregators Module (Aggregator Module)\]**​ をクリックします。 4. **\[Add (追加)\]**​ をクリックします。 5. **\[Finish (完了)\]**​ をクリックします。 Studio でモジュールを Mule プロジェクトに追加しても、Studio ワークスペースの他のプロジェクトはそのモジュールを使用できません。 ## [](#configure-source)ソースの設定 ソースは、指定された条件が満たされたときにフローを開始します。 次のソースのいずれかを Aggregators Module で使用するように設定できます。 - **\[HTTP\] > \[Listener\]** 設定されたホストとポートで要求を受信するたびにフローを開始します。 - **Aggregator listener (アグリゲーターリスナー)** 参照されるアグリゲータースコープによりトリガーされる要素をリスンすることで、サブフローを開始します。 リスナーで参照されるアグリゲータースコープが集約を完了すると、​**\[Aggregator listener (アグリゲーターリスナー)\]**​ はすべての集約された要素のリストを使用してサブフローをトリガーします。 たとえば、\[Aggregator listener (アグリゲーターリスナー)\] を設定する手順は、次のとおりです。 1. **\[Mule Palette (Mule パレット)\]**​ ビューで、​**\[Aggregators (アグリゲーター)\] > \[Aggregator listener (アグリゲーターリスナー)\]**​ を選択します。 2. **\[Aggregator listener (アグリゲーターリスナー)\]**​ を Studio キャンバスにドラッグします。 3. **\[Aggregator listener (アグリゲーターリスナー)\]**​ 設定画面で、必要に応じて ​**\[Display Name (表示名)\]**​ 項目の値を変更します。 4. **\[Aggregator name (アグリゲーター名)\]**​ を、リスンするアグリゲータースコープの名前に設定します。 アグリゲータースコープがその要素を解放すると、リスナーが実行されます。 5. **\[Include timed out groups (タイムアウトグループを含める)\]**​ を選択し、タイムアウトによりグループが解放されたときに、リスナーがトリガーされるかどうかを示します。 この設定オプションは、​**\[Time based aggregator (時間ベースのアグリゲーター)\]**​ スコープや ​**\[Size based aggregator (サイズベースのアグリゲーター)\]**​ スコープなど、グループ以外のアグリゲーターでも機能します。 ## [](#add-connector-operation)アグリゲータースコープをフローに追加する フローにモジュールスコープを追加するときは、そのモジュールで実行するアクションを指定しています。 Aggregators Module のスコープを追加する手順は、次のとおりです。 1. **\[Mule Palette (Mule パレット)\]**​ ビューで ​**\[Aggregators (アグリゲーター)\]**​ を選択し、目的のスコープを選択します。 - **Group based aggregator (グループベースのアグリゲーター)** 要素をグループ ID でグループに集約できます。 - **Size based aggregator (サイズベースのアグリゲーター)** 定義済みのサイズ数の要素の集約が完了するまで要素を集約することができます。 - **Time based aggregator (時間ベースのアグリゲーター)** 定義済みの時間制限内で要素を集約できます。 2. スコープを Studio キャンバスのソースの右にドラッグします。 ![Anypoint Studio の \[Mule Palette (Mule パレット)\] ビュー内の Aggregator Module スコープとソース](_images/aggregators-scopes.png) Figure 1. Studio での Aggregator Module スコープ ## [](#configure-os-aggregator)アグリゲーターのオブジェクトストアを設定する どのアグリゲータースコープでも、グローバルオブジェクトストアを参照するか非公開オブジェクトストアを作成することで、オブジェクトストアを設定できます。 ### [](#オブジェクトストアを設定する)オブジェクトストアを設定する Studio でグローバルオブジェクトストアを設定するには、次の手順に従います。 1. フローからアグリゲータースコープを選択します。 2. アグリゲータースコープ設定画面で、​**\[Advanced (詳細)\]**​ をクリックします。 3. 緑のプラス記号 (+) をクリックして、グローバルオブジェクトストア設定を作成します。 4. **\[Global Object store configuration (グローバルオブジェクトストア設定)\]**​ 画面で、​**\[Name (名前)\]**​ を ​`aGlobalObjectStore`​ に設定します。 5. 必要な設定属性を入力して、​**\[OK\]**​ をクリックします。 次のスクリーンショットは、​**\[Group based aggregator (グループベースアグリゲーター)\]**​ スコープの ​**\[Advanced (詳細)\]**​ タブのオブジェクトストア設定を示しています。 ![「Object Store(\*) (オブジェクトストア(\*))」 項目では aGlobalObjectStore 設定が選択されています。](/mulesoft/current/ja-jp/aggregators-module/latest/_images/aggregators-os-global.png) **\[Configuration XML (設定 XML)\]**​ ビューでは ​`objectStore`​ 設定は次のように表示されます。 ```xml ``` ### [](#非公開オブジェクトストアの設定)非公開オブジェクトストアの設定 Studio で非公開オブジェクトストアを設定するには、XML コードを手動で更新します。 1. フローからアグリゲータースコープを選択します。 2. Studio キャンバスで ​**\[Configuration XML (設定 XML)\]**​ をクリックします。 3. XML の非公開オブジェクトストア ​``​ を ​``​ セクションの内側に追加します。次に例を示します。 ```xml ... ``` ## [](#view-app-log)アプリケーションログの表示 問題を確認するため、アプリケーションログを次の方法で表示できます。 - アプリケーションを Anypoint Platform から実行している場合、アプリケーションログ出力は Anypoint Studio のコンソールウィンドウに表示されます。 - コマンドラインから Mule を使用してアプリケーションを実行している場合、アプリケーションログ出力は OS コンソールに表示されます。 アプリケーションのログファイル (​`log4j2.xml`​) でログファイルパスがカスタマイズされていない場合、デフォルトの場所 ​`MULE_HOME/logs/.log`​ でアプリケーションログを表示することもできます。 ## [](#関連情報)関連情報 - [Anypoint Connector の概要](https://docs.mulesoft.com/connectors/introduction/introduction-to-anypoint-connectors) - [Anypoint Studio を使用したコネクタの設定](https://docs.mulesoft.com/connectors/introduction/intro-config-use-studio) - [Aggregators Module リファレンス](https://docs.mulesoft.com/jp/aggregators-module/latest/aggregators-module-reference) - [MuleSoft ヘルプセンター](https://help.mulesoft.com) # クラスターでのアグリゲーターの使用
CloudHub ではクラスタリングを使用できないため、アグリゲーターは複数のワーカーおよび永続的なオブジェクトストアでは機能しません。
Aggregators Module は、クラスターの標準環境で動作します。ただし、予期しない動作を回避するため、以下の点を考慮してください。 非同期集約が定義されている場合、最初の要素が到着したときに、その要素はクラスターのプライマリノードでスケジュールされます。新しい値はクラスターのどのノードにも到着する可能性があるため、通知してプライマリノードでその集約をスケジュールさせる方法が必要です。そのために、プライマリノードの別のタスクは、新しい集約をスケジュールする必要があるかどうかを定期的に判断します。これは、新しい集約スケジュールをチェックする間隔が集約の実際のタイムアウトよりもはるかに長い場合に問題になる可能性があります。この理由は、集約がスケジュールされる前に終了したり、時間の計算でエラーが発生する可能性があるためです。 この問題を回避するため、スケジュールする新しい集約があるかどうかをプライマリノードがチェックする頻度を設定します。次のいずれかを使用してこの値を定義します。 - グローバル設定プロパティ ​`mule.aggregatorsSchedulingPeriod`​ (ミリ秒) - システムプロパティ ​`-M-Dmule.aggregatorsSchedulingPeriod` グローバルプロパティまたはシステムプロパティの設定方法についての詳細は、​[「プロパティの設定」](https://docs.mulesoft.com/mule-runtime/latest/configuring-properties)​を参照してください。 ## [](#関連情報)関連情報 - [Aggregators Module リファレンス](https://docs.mulesoft.com/jp/aggregators-module/1.0/aggregators-module-reference) - [MuleSoft ヘルプセンター](https://help.mulesoft.com) # クラスターでのアグリゲーターの使用
CloudHub ではクラスタリングを使用できないため、アグリゲーターは複数のワーカーおよび永続的なオブジェクトストアでは機能しません。
Aggregators Module は、クラスターの標準環境で動作します。ただし、予期しない動作を回避するため、以下の点を考慮してください。 非同期集約が定義されている場合、最初の要素が到着したときに、その要素はクラスターのプライマリノードでスケジュールされます。新しい値はクラスターのどのノードにも到着する可能性があるため、通知してプライマリノードでその集約をスケジュールさせる方法が必要です。そのために、プライマリノードの別のタスクは、新しい集約をスケジュールする必要があるかどうかを定期的に判断します。これは、新しい集約スケジュールをチェックする間隔が集約の実際のタイムアウトよりもはるかに長い場合に問題になる可能性があります。この理由は、集約がスケジュールされる前に終了したり、時間の計算でエラーが発生する可能性があるためです。 この問題を回避するため、スケジュールする新しい集約があるかどうかをプライマリノードがチェックする頻度を設定します。次のいずれかを使用してこの値を定義します。 - グローバル設定プロパティ ​`mule.aggregatorsSchedulingPeriod`​ (ミリ秒) - システムプロパティ ​`-M-Dmule.aggregatorsSchedulingPeriod` グローバルプロパティまたはシステムプロパティの設定方法についての詳細は、​[「プロパティの設定」](https://docs.mulesoft.com/mule-runtime/latest/configuring-properties)​を参照してください。 ## [](#関連情報)関連情報 - [Aggregators Module リファレンス](https://docs.mulesoft.com/jp/aggregators-module/latest/aggregators-module-reference) - [MuleSoft ヘルプセンター](https://help.mulesoft.com) # グループベースのアグリゲーターの例 Aggregators Module の ​**\[Group based aggregator (グループベースのアグリゲーター)\]**​ スコープを使用すると、要素をグループ ID でグループに集約できます。次の例は、要求で送信されたグループ ID に基づいて、グループごとに最大 ​`3`​ 件の HTTP 要求コールを集約するようにこのスコープを設定する方法を示しています。アプリケーションは DataWeave を使用して、ランダム ID 番号で各コールのペイロードを設定します。各グループで ​`3`​ 個の要素の集約が完了すると、アプリケーションはグループのすべての集約されたランダム ID 番号要素を記録します。 ![Anypoint Studio キャンバスでのグループベースのアグリゲーターのフローの例](/mulesoft/current/ja-jp/aggregators-module/1.0/_images/aggregator-group-example.png) Figure 1. グループベースのアグリゲーターのフローの例 独自の環境でこの例をテストするには、Mule アプリケーションを作成し、curl コマンドを使用してアプリケーションを実行する必要があります。 ## [](#mule-アプリケーションの作成)Mule アプリケーションの作成 Anypoint Studio で Mule フローを作成する手順は、次のとおりです。 1. **\[Mule Palette (Mule パレット)\]**​ で、​**\[HTTP Listener (HTTP リスナー)\]**​ を選択してキャンバスにドラッグします。 このソースは受信 HTTP メッセージ属性をリスンすることでフローを開始します。 2. **\[Connector configuration (コネクタ設定)\]**​ で ​**\[HTTP\_Listener\_config\]**​ を選択します。 3. **\[Path (パス)\]**​ を ​`/test`​ に設定します。 4. **\[Transform message\]**​ コンポーネントを ​**\[Listener\]**​ の右にドラッグします。 5. **\[Transform message\]**​ コンポーネントの DataSense プレビューウィンドウで、次の DataWeave 式を追加します。これにより、各 HTTP 要求のペイロードに ​`Source Name`​ が ​`"group-based"`​ として設定され、​`Source ID`​ が ​`random() as String`​ として設定されます。また、​`message.attributes['queryParams'].groupId`​ に基づいて要求の ​`GroupId`​ 値も取得します。 ```DataWeave %dw 2.0 output application/json --- { "Source Name": "group-based", "GroupId": message.attributes['queryParams'].groupId, "Source ID": (random() as String) } ``` 6. **\[Group based aggregator (グループベースのアグリゲーター)\]**​ スコープを ​**\[Transform message\]**​ の右にドラッグします。 7. **\[名前\]**​ を ​`groupBasedAggregator`​ に設定します。 8. **\[Content (コンテンツ)\]**​ を ​`payload`​ に設定します。これは、何を集約するか (この場合は変換された HTTP 要求) を定義する式です。 9. **\[Group id (グループ ID)\]**​ を ​`payload.GroupId`​ に設定します。これは、受信した各新規メッセージで評価される式であり、要素を集約するグループ ID を取得します。 10. **\[Group size (グループサイズ)\]**​ を ​`3`​ に設定します。これは、解決したグループ ID を使用してグループに集約する要素の最大サイズです。 11. **\[Logger\]**​ コンポーネントを ​**\[Incremental aggregation (増分集約)\]**​ ルートにドラッグします。 12. **\[Logger\]**​ 設定画面で、​**\[Message (メッセージ)\]**​ を ​`Doing incremental aggregation.`​ に設定します。 13. **\[Logger\]**​ コンポーネントを ​**\[Aggregation complete (集約完了)\]**​ ルートにドラッグします。 14. **\[Message (メッセージ)\]**​ を ​`Aggregation complete`​ に設定します。 15. 別の ​**\[Logger\]**​ コンポーネントを ​**\[Aggregation complete (集約完了)\]**​ 内の最初の ​**\[Logger\]**​ コンポーネントの右にドラッグします。 16. **\[Message (メッセージ)\]**​ を次の式に設定します。この式では、すべての集約された要素が返されます。 ```xml output application/json --- payload ``` 17. プロジェクトを保存します。 18. **Package Explorer**​ で、​**\[Run (実行)\] > \[Run As (別のユーザーとして実行)\] > \[Mule Application (Mule アプリケーション)\]**​ をクリックします。 19. 次の curl コマンドを 2 回送信して、アプリケーションをテストします: + ​`curl -X POST [http://localhost:8081/test?groupId=1](http://localhost:8081/test?groupId=1)`​。 これにより、2 つの HTTP 要求が生成され、グループ ID ​`1`​ に集約されます。 20. コマンド ​`curl -X POST [http://localhost:8081/test?groupId=2](http://localhost:8081/test?groupId=2)`​ を送信して、1 つの要求をグループ ID ​`2`​ に集約します。 21. 別のコマンド ​`curl -X POST [http://localhost:8081/test?groupId=1](http://localhost:8081/test?groupId=1)`​ を送信して、要求をグループ ID ​`1`​ に集約します。 22. **\[Console (コンソール)\]**​ ビューに移動して、ロガーメッセージを参照します。 ```none INFO 2021-06-14 16:34:33,921 [[MuleRuntime].uber.12: [aggregator-group-demo].aggregator-group-demoFlow.CPU_INTENSIVE @2c3478b5] [processor: aggregator-group-demoFlow/processors/1/route/0/processors/0; event: 8beae010-cd47-11eb-8b82-f01898ad2638] org.mule.runtime.core.internal.processor.LoggerMessageProcessor: Doing incremental aggregation INFO 2021-06-14 16:34:43,618 [[MuleRuntime].uber.12: [aggregator-group-demo].aggregator-group-demoFlow.CPU_INTENSIVE @2c3478b5] [processor: aggregator-group-demoFlow/processors/1/route/0/processors/0; event: 91b93be0-cd47-11eb-8b82-f01898ad2638] org.mule.runtime.core.internal.processor.LoggerMessageProcessor: Doing incremental aggregation INFO 2021-06-14 16:34:47,203 [[MuleRuntime].uber.12: [aggregator-group-demo].aggregator-group-demoFlow.CPU_INTENSIVE @2c3478b5] [processor: aggregator-group-demoFlow/processors/1/route/0/processors/0; event: 93d8e790-cd47-11eb-8b82-f01898ad2638] org.mule.runtime.core.internal.processor.LoggerMessageProcessor: Doing incremental aggregation INFO 2021-06-14 16:34:51,380 [[MuleRuntime].uber.12: [aggregator-group-demo].aggregator-group-demoFlow.CPU_INTENSIVE @2c3478b5] [processor: aggregator-group-demoFlow/processors/1/route/1/processors/0; event: 96566ab0-cd47-11eb-8b82-f01898ad2638] org.mule.runtime.core.internal.processor.LoggerMessageProcessor: Aggregation complete INFO 2021-06-14 16:34:51,409 [[MuleRuntime].uber.12: [aggregator-group-demo].aggregator-group-demoFlow.CPU_INTENSIVE @2c3478b5] [processor: aggregator-group-demoFlow/processors/1/route/1/processors/1; event: 96566ab0-cd47-11eb-8b82-f01898ad2638] org.mule.runtime.core.internal.processor.LoggerMessageProcessor: [ { "Source Name": "group-based", "GroupId": "1", "Source ID": "0.5127046416902801" }, { "Source Name": "group-based", "GroupId": "1", "Source ID": "0.20239307122656847" }, { "Source Name": "group-based", "GroupId": "1", "Source ID": "0.7505233647798346" } ] ``` このロガーメッセージの例は、グループ ID ​`1`​ の最初の 2 つの HTTP 要求とグループ ID ​`2`​ の 3 番目の要求で、​**\[Incremental aggregator (増分アグリゲーター)\]**​ ルートの ​**\[Logger\]**​ コンポーネントによってメッセージ ​`Doing incremental aggregation.`​ が記録されることを示しています。 4 番目の HTTP 要求 (グループ ID ​`1`​ の 3 番目の要求) を送信すると、グループサイズがグループあたり ​`3`​ 件のコールになっているため、集約グループ ​`1`​ は完了します。​**\[Aggregation complete (集約完了)\]**​ ルートの 1 番目の ​**\[Logger\]**​ コンポーネントによってメッセージ ​`Aggregation complete`​ が記録され、2 番目の ​**\[Logger\]**​ コンポーネントによってグループの集約されたペイロード要素 (ランダム ID 番号) が返されます。 コールをグループ ​`2`​ に集約する別の 2 つのコマンド ​`curl -X POST [http://localhost:8081/test?groupId=2](http://localhost:8081/test?groupId=2)`​ を送信すると、​**\[Console (コンソール)\]**​ ビューに次のロガーメッセージが返されます。 ```none INFO 2021-06-14 16:53:56,262 [[MuleRuntime].uber.12: [aggregator-group-demo].aggregator-group-demoFlow.CPU_INTENSIVE @2c3478b5] [processor: aggregator-group-demoFlow/processors/1/route/0/processors/0; event: 40c07200-cd4a-11eb-8b82-f01898ad2638] org.mule.runtime.core.internal.processor.LoggerMessageProcessor: Doing incremental aggregation INFO 2021-06-14 16:53:58,686 [[MuleRuntime].uber.01: [aggregator-group-demo].aggregator-group-demoFlow.CPU_INTENSIVE @2c3478b5] [processor: aggregator-group-demoFlow/processors/1/route/1/processors/0; event: 42329fa0-cd4a-11eb-8b82-f01898ad2638] org.mule.runtime.core.internal.processor.LoggerMessageProcessor: Aggregation complete INFO 2021-06-14 16:53:58,689 [[MuleRuntime].uber.01: [aggregator-group-demo].aggregator-group-demoFlow.CPU_INTENSIVE @2c3478b5] [processor: aggregator-group-demoFlow/processors/1/route/1/processors/1; event: 42329fa0-cd4a-11eb-8b82-f01898ad2638] org.mule.runtime.core.internal.processor.LoggerMessageProcessor: [ { "Source Name": "group-based", "GroupId": "2", "Source ID": "0.8390326267445475" }, { "Source Name": "group-based", "GroupId": "2", "Source ID": "0.8931962767228987" }, { "Source Name": "group-based", "GroupId": "2", "Source ID": "0.6514448303865373" } ] ``` ## [](#グループベースのアグリゲーターの例の-xml)グループベースのアグリゲーターの例の XML この例のフローをすばやく Mule アプリケーションに読み込むには、次のコードを Studio XML エディターに貼り付けます。 ```xml ``` ## [](#関連情報)関連情報 - [Aggregators Module の例](https://docs.mulesoft.com/jp/aggregators-module/1.0/aggregators-examples) - [MuleSoft ヘルプセンター](https://help.mulesoft.com) # グループベースのアグリゲーターの例 Aggregators Module の ​**\[Group based aggregator (グループベースのアグリゲーター)\]**​ スコープを使用すると、要素をグループ ID でグループに集約できます。次の例は、要求で送信されたグループ ID に基づいて、グループごとに最大 ​`3`​ 件の HTTP 要求コールを集約するようにこのスコープを設定する方法を示しています。アプリケーションは DataWeave を使用して、ランダム ID 番号で各コールのペイロードを設定します。各グループで ​`3`​ 個の要素の集約が完了すると、アプリケーションはグループのすべての集約されたランダム ID 番号要素を記録します。 ![Anypoint Studio キャンバスでのグループベースのアグリゲーターのフローの例](/mulesoft/current/ja-jp/aggregators-module/latest/_images/aggregator-group-example.png) Figure 1. グループベースのアグリゲーターのフローの例 独自の環境でこの例をテストするには、Mule アプリケーションを作成し、curl コマンドを使用してアプリケーションを実行する必要があります。 ## [](#mule-アプリケーションの作成)Mule アプリケーションの作成 Anypoint Studio で Mule フローを作成する手順は、次のとおりです。 1. **\[Mule Palette (Mule パレット)\]**​ で、​**\[HTTP Listener (HTTP リスナー)\]**​ を選択してキャンバスにドラッグします。 このソースは受信 HTTP メッセージ属性をリスンすることでフローを開始します。 2. **\[Connector configuration (コネクタ設定)\]**​ で ​**\[HTTP\_Listener\_config\]**​ を選択します。 3. **\[Path (パス)\]**​ を ​`/test`​ に設定します。 4. **\[Transform message\]**​ コンポーネントを ​**\[Listener\]**​ の右にドラッグします。 5. **\[Transform message\]**​ コンポーネントの DataSense プレビューウィンドウで、次の DataWeave 式を追加します。これにより、各 HTTP 要求のペイロードに ​`Source Name`​ が ​`"group-based"`​ として設定され、​`Source ID`​ が ​`random() as String`​ として設定されます。また、​`message.attributes['queryParams'].groupId`​ に基づいて要求の ​`GroupId`​ 値も取得します。 ```DataWeave %dw 2.0 output application/json --- { "Source Name": "group-based", "GroupId": message.attributes['queryParams'].groupId, "Source ID": (random() as String) } ``` 6. **\[Group based aggregator (グループベースのアグリゲーター)\]**​ スコープを ​**\[Transform message\]**​ の右にドラッグします。 7. **\[名前\]**​ を ​`groupBasedAggregator`​ に設定します。 8. **\[Content (コンテンツ)\]**​ を ​`payload`​ に設定します。これは、何を集約するか (この場合は変換された HTTP 要求) を定義する式です。 9. **\[Group id (グループ ID)\]**​ を ​`payload.GroupId`​ に設定します。これは、受信した各新規メッセージで評価される式であり、要素を集約するグループ ID を取得します。 10. **\[Group size (グループサイズ)\]**​ を ​`3`​ に設定します。これは、解決したグループ ID を使用してグループに集約する要素の最大サイズです。 11. **\[Logger\]**​ コンポーネントを ​**\[Incremental aggregation (増分集約)\]**​ ルートにドラッグします。 12. **\[Logger\]**​ 設定画面で、​**\[Message (メッセージ)\]**​ を ​`Doing incremental aggregation.`​ に設定します。 13. **\[Logger\]**​ コンポーネントを ​**\[Aggregation complete (集約完了)\]**​ ルートにドラッグします。 14. **\[Message (メッセージ)\]**​ を ​`Aggregation complete`​ に設定します。 15. 別の ​**\[Logger\]**​ コンポーネントを ​**\[Aggregation complete (集約完了)\]**​ 内の最初の ​**\[Logger\]**​ コンポーネントの右にドラッグします。 16. **\[Message (メッセージ)\]**​ を次の式に設定します。この式では、すべての集約された要素が返されます。 ```xml output application/json --- payload ``` 17. プロジェクトを保存します。 18. **Package Explorer**​ で、​**\[Run (実行)\] > \[Run As (別のユーザーとして実行)\] > \[Mule Application (Mule アプリケーション)\]**​ をクリックします。 19. 次の curl コマンドを 2 回送信して、アプリケーションをテストします: + ​`curl -X POST [http://localhost:8081/test?groupId=1](http://localhost:8081/test?groupId=1)`​。 これにより、2 つの HTTP 要求が生成され、グループ ID ​`1`​ に集約されます。 20. コマンド ​`curl -X POST [http://localhost:8081/test?groupId=2](http://localhost:8081/test?groupId=2)`​ を送信して、1 つの要求をグループ ID ​`2`​ に集約します。 21. 別のコマンド ​`curl -X POST [http://localhost:8081/test?groupId=1](http://localhost:8081/test?groupId=1)`​ を送信して、要求をグループ ID ​`1`​ に集約します。 22. **\[Console (コンソール)\]**​ ビューに移動して、ロガーメッセージを参照します。 ```none INFO 2021-06-14 16:34:33,921 [[MuleRuntime].uber.12: [aggregator-group-demo].aggregator-group-demoFlow.CPU_INTENSIVE @2c3478b5] [processor: aggregator-group-demoFlow/processors/1/route/0/processors/0; event: 8beae010-cd47-11eb-8b82-f01898ad2638] org.mule.runtime.core.internal.processor.LoggerMessageProcessor: Doing incremental aggregation INFO 2021-06-14 16:34:43,618 [[MuleRuntime].uber.12: [aggregator-group-demo].aggregator-group-demoFlow.CPU_INTENSIVE @2c3478b5] [processor: aggregator-group-demoFlow/processors/1/route/0/processors/0; event: 91b93be0-cd47-11eb-8b82-f01898ad2638] org.mule.runtime.core.internal.processor.LoggerMessageProcessor: Doing incremental aggregation INFO 2021-06-14 16:34:47,203 [[MuleRuntime].uber.12: [aggregator-group-demo].aggregator-group-demoFlow.CPU_INTENSIVE @2c3478b5] [processor: aggregator-group-demoFlow/processors/1/route/0/processors/0; event: 93d8e790-cd47-11eb-8b82-f01898ad2638] org.mule.runtime.core.internal.processor.LoggerMessageProcessor: Doing incremental aggregation INFO 2021-06-14 16:34:51,380 [[MuleRuntime].uber.12: [aggregator-group-demo].aggregator-group-demoFlow.CPU_INTENSIVE @2c3478b5] [processor: aggregator-group-demoFlow/processors/1/route/1/processors/0; event: 96566ab0-cd47-11eb-8b82-f01898ad2638] org.mule.runtime.core.internal.processor.LoggerMessageProcessor: Aggregation complete INFO 2021-06-14 16:34:51,409 [[MuleRuntime].uber.12: [aggregator-group-demo].aggregator-group-demoFlow.CPU_INTENSIVE @2c3478b5] [processor: aggregator-group-demoFlow/processors/1/route/1/processors/1; event: 96566ab0-cd47-11eb-8b82-f01898ad2638] org.mule.runtime.core.internal.processor.LoggerMessageProcessor: [ { "Source Name": "group-based", "GroupId": "1", "Source ID": "0.5127046416902801" }, { "Source Name": "group-based", "GroupId": "1", "Source ID": "0.20239307122656847" }, { "Source Name": "group-based", "GroupId": "1", "Source ID": "0.7505233647798346" } ] ``` このロガーメッセージの例は、グループ ID ​`1`​ の最初の 2 つの HTTP 要求とグループ ID ​`2`​ の 3 番目の要求で、​**\[Incremental aggregator (増分アグリゲーター)\]**​ ルートの ​**\[Logger\]**​ コンポーネントによってメッセージ ​`Doing incremental aggregation.`​ が記録されることを示しています。 4 番目の HTTP 要求 (グループ ID ​`1`​ の 3 番目の要求) を送信すると、グループサイズがグループあたり ​`3`​ 件のコールになっているため、集約グループ ​`1`​ は完了します。​**\[Aggregation complete (集約完了)\]**​ ルートの 1 番目の ​**\[Logger\]**​ コンポーネントによってメッセージ ​`Aggregation complete`​ が記録され、2 番目の ​**\[Logger\]**​ コンポーネントによってグループの集約されたペイロード要素 (ランダム ID 番号) が返されます。 コールをグループ ​`2`​ に集約する別の 2 つのコマンド ​`curl -X POST [http://localhost:8081/test?groupId=2](http://localhost:8081/test?groupId=2)`​ を送信すると、​**\[Console (コンソール)\]**​ ビューに次のロガーメッセージが返されます。 ```none INFO 2021-06-14 16:53:56,262 [[MuleRuntime].uber.12: [aggregator-group-demo].aggregator-group-demoFlow.CPU_INTENSIVE @2c3478b5] [processor: aggregator-group-demoFlow/processors/1/route/0/processors/0; event: 40c07200-cd4a-11eb-8b82-f01898ad2638] org.mule.runtime.core.internal.processor.LoggerMessageProcessor: Doing incremental aggregation INFO 2021-06-14 16:53:58,686 [[MuleRuntime].uber.01: [aggregator-group-demo].aggregator-group-demoFlow.CPU_INTENSIVE @2c3478b5] [processor: aggregator-group-demoFlow/processors/1/route/1/processors/0; event: 42329fa0-cd4a-11eb-8b82-f01898ad2638] org.mule.runtime.core.internal.processor.LoggerMessageProcessor: Aggregation complete INFO 2021-06-14 16:53:58,689 [[MuleRuntime].uber.01: [aggregator-group-demo].aggregator-group-demoFlow.CPU_INTENSIVE @2c3478b5] [processor: aggregator-group-demoFlow/processors/1/route/1/processors/1; event: 42329fa0-cd4a-11eb-8b82-f01898ad2638] org.mule.runtime.core.internal.processor.LoggerMessageProcessor: [ { "Source Name": "group-based", "GroupId": "2", "Source ID": "0.8390326267445475" }, { "Source Name": "group-based", "GroupId": "2", "Source ID": "0.8931962767228987" }, { "Source Name": "group-based", "GroupId": "2", "Source ID": "0.6514448303865373" } ] ``` ## [](#グループベースのアグリゲーターの例の-xml)グループベースのアグリゲーターの例の XML この例のフローをすばやく Mule アプリケーションに読み込むには、次のコードを Studio XML エディターに貼り付けます。 ```xml ``` ## [](#関連情報)関連情報 - [Aggregators Module の例](https://docs.mulesoft.com/jp/aggregators-module/latest/aggregators-examples) - [MuleSoft ヘルプセンター](https://help.mulesoft.com) # サイズベースのアグリゲーターのタイムアウト項目と最大サイズ項目の設定 Aggregators Module の ​**\[Size based aggregator (サイズベースのアグリゲーター)\]**​ スコープのタイムアウト動作を理解するには、​**\[Timeout (タイムアウト)\]**​ 項目と ​**\[Max size (最大サイズ)\]**​ 項目のしくみについて説明している次の図を参照してください。その後、Mule アプリケーションの例でこれらの項目を Anypoint Studio と XML エディターで設定する方法を確認してください。 ![aggregator size timeout diagram](/mulesoft/current/ja-jp/aggregators-module/1.0/_images/aggregator-size-timeout-diagram.png) Figure 1. サイズベースのアグリゲーターのタイムアウトと最大サイズの動作の図 - **\[Max size (最大サイズ)\]**​ 項目は、集約が完了とみなされる前に集約される要素の合計数を示します。 - **\[Timeout (タイムアウト)\]**​ 項目は集約が完了するのを待機する最大時間を示します。 - 最初のプロセスは、タイムアウトの発生前に集約された要素の合計数が必要な数に達したために集約が完了したことを示しています。 - 2 番目のプロセスは、タイムアウトの発生前に集約された要素の合計数が必要な数に達しなかったために集約が完了していないことを示しています。これにより、集約はリセットされます。 - 3 番目のプロセスは、​**\[Aggregator listener (アグリゲーターリスナー)\]**​ ソースを使用して、タイムアウトの発生前に集約された要素を取得できることを示しています。このソースは、タイムアウトが発生するとトリガーされます。 **\[Aggregator listener (アグリゲーターリスナー)\]**​ ソースを使用して未完了の集約要素を取得する場合、​**\[Include timed out group (タイムアウトグループを含める)\]**​ 項目を選択して、タイムアウトの発生時にソースがトリガーされることを示す必要があります。 ## [](#mule-アプリケーションの設定)Mule アプリケーションの設定 次の例は、​`10`​ 秒間に最大 ​`3`​ 件の HTTP 要求コールを集約するように ​**\[Size based Aggregator (サイズベースのアグリゲーター)\]**​ スコープを設定する方法を示しています。アプリケーションは DataWeave を使用して、ランダム ID 番号で各コールのペイロードを設定します。タイムアウトの発生前に集約の完了に必要な ​`3`​ 件のコールに達すると、アプリケーションはすべての集約された要素を記録します。それ以外の場合、アプリケーションは、タイムアウトの発生時に ​**\[Aggregator listener (アグリゲーターリスナー)\]**​ ソースによって取得された集約済み要素のみを記録します。 ![Anypoint Studio キャンバスでのサイズベースのアグリゲーターのフローの例](/mulesoft/current/ja-jp/aggregators-module/1.0/_images/aggregator-size-timeout-example.png) Figure 2. サイズベースのアグリゲーターのフローの例 独自の環境でこの例をテストするには、Mule アプリケーションを作成し、curl コマンドを使用してアプリケーションを実行する必要があります。 Anypoint Studio で Mule フローを作成する手順は、次のとおりです。 1. **\[Mule Palette (Mule パレット)\]**​ で、​**\[HTTP Listener (HTTP リスナー)\]**​ を選択してキャンバスにドラッグします。 このソースは受信 HTTP メッセージ属性をリスンすることでフローを開始します。 2. **\[Connector configuration (コネクタ設定)\]**​ で ​**\[HTTP\_Listener\_config\]**​ を選択します。 3. **\[Path (パス)\]**​ を ​`/test`​ に設定します。 4. **\[Transform message\]**​ コンポーネントを ​**\[Listener\]**​ の右にドラッグします。 5. **\[Transform message\]**​ コンポーネントの DataSense プレビューウィンドウで、次の DataWeave 式を追加します。これにより、各 HTTP 要求のペイロードに ​`Source Name`​ が ​`"size-based"`​ として設定され、​`Source ID`​ が ​`random() as String`​ として設定されます。 ```DataWeave %dw 2.0 output application/json --- { "Source Name": "size-based", "Source ID": (random() as String) } ``` 6. **\[Size based aggregator (サイズベースのアグリゲーター)\]**​ スコープを ​**\[Transform message\]**​ の右にドラッグします。 7. **\[名前\]**​ を ​`sizeBasedAggregator`​ に設定します。 8. **\[Max Size (最大サイズ)\]**​ を ​`3`​ に設定します。 9. **\[Timeout (タイムアウト)\]**​ を ​`10`​ に設定します。 10. **\[Content (コンテンツ)\]**​ を ​`payload`​ に設定します。これは、何を集約するか (この場合は変換された HTTP 要求) を定義する式です。 11. **\[Logger\]**​ コンポーネントを ​**\[Incremental aggregation (増分集約)\]**​ ルートにドラッグします。 12. **\[Logger\]**​ 設定画面で、​**\[Message (メッセージ)\]**​ を ​`Doing incremental aggregation step.`​ に設定します。 13. **\[Logger\]**​ コンポーネントを ​**\[Aggregation complete (集約完了)\]**​ ルートにドラッグします。 14. **\[Message (メッセージ)\]**​ を ​`Aggregation complete`​ に設定します。 15. 別の ​**\[Logger\]**​ コンポーネントを ​**\[Aggregation complete (集約完了)\]**​ 内の最初の ​**\[Logger\]**​ コンポーネントの右にドラッグします。 16. **\[Message (メッセージ)\]**​ を次の式に設定します。この式では、すべての集約された要素が返されます。 ```xml output application/json --- payload ``` 17. **\[Aggregator listener (アグリゲーターリスナー)\]**​ ソースを最初のフローの下にドラッグします。 18. **\[Aggregator name (アグリゲーター名)\]**​ を ​`sizeBasedAggregator`​ に設定します。これは、前のアグリゲーターをリスンします。 19. **\[SelectInclude Timed Out Groups (タイムアウトグループを含める)\]**​ を選択します。これは、タイムアウトの発生時にリスナーがトリガーされるかどうかを示します。 20. **\[Logger\]**​ コンポーネントを ​**\[Aggregation listener (集約リスナー)\]**​ ソースの右にドラッグします。 21. **\[Message (メッセージ)\]**​ を次の式に設定します。この式では、タイムアウトの発生前に集約された要素のみが返されます。 ```xml output application/json --- payload ``` 22. プロジェクトを保存します。 23. **Package Explorer**​ で、​**\[Run (実行)\] > \[Run As (別のユーザーとして実行)\] > \[Mule Application (Mule アプリケーション)\]**​ をクリックします。 24. 10 秒以内に次の curl コマンドを送信して、アプリケーションをテストします。 `curl -X POST [http://localhost:8086/test](http://localhost:8086/test)` 25. **\[Console (コンソール)\]**​ ビューに移動して、ロガーメッセージを参照します。 ```none INFO 2021-06-22 12:59:16,319 [[MuleRuntime].uber.12: [aggregator-size-demo].aggregator-size-demoFlow.CPU_INTENSIVE @7992f689] [processor: aggregator-size-demoFlow/processors/1/route/0/processors/0; event: cbc219c0-d372-11eb-b975-f01898ad2638] org.mule.runtime.core.internal.processor.LoggerMessageProcessor: Doing incremental aggregation step. INFO 2021-06-22 12:59:26,328 [[MuleRuntime].uber.12: [aggregator-size-demo].aggregator-size-demoFlow1.CPU_LITE @684a1bef] [processor: aggregator-size-demoFlow1/processors/0; event: d6b48f00-d36f-11eb-b975-f01898ad2638] org.mule.runtime.core.internal.processor.LoggerMessageProcessor: [ { "Source Name": "size-based", "Source ID": "0.6568747259174192" } ] ``` このロガーメッセージの例は、最初の HTTP 要求で、​**\[Incremental aggregator (増分アグリゲーター)\]**​ ルートの ​**\[Logger\]**​ コンポーネントによってメッセージ ​`Doing incremental aggregation step.`​ が記録されることを示しています。集約が ​`10`​ 秒に達すると、タイムアウトが発生します。​**\[Aggregation listener (集約リスナー)\]**​ ソースによって唯一の HTTP 要求コールが取得され、後続の ​**\[Logger\]**​ コンポーネントによって、集約されたペイロード要素 (ランダム ID 番号) が返されます。 ## [](#サイズベースのアグリゲーターのタイムアウトと最大サイズの例の-xml)サイズベースのアグリゲーターのタイムアウトと最大サイズの例の XML この例のフローをすばやく Mule アプリケーションに読み込むには、次のコードを Studio XML エディターに貼り付けます。 ```xml ``` ## [](#関連情報)関連情報 - [Aggregators Module の例](https://docs.mulesoft.com/jp/aggregators-module/1.0/aggregators-examples) - [MuleSoft ヘルプセンター](https://help.mulesoft.com) # サイズベースのアグリゲーターのタイムアウト項目と最大サイズ項目の設定 Aggregators Module の ​**\[Size based aggregator (サイズベースのアグリゲーター)\]**​ スコープのタイムアウト動作を理解するには、​**\[Timeout (タイムアウト)\]**​ 項目と ​**\[Max size (最大サイズ)\]**​ 項目のしくみについて説明している次の図を参照してください。その後、Mule アプリケーションの例でこれらの項目を Anypoint Studio と XML エディターで設定する方法を確認してください。 ![aggregator size timeout diagram](/mulesoft/current/ja-jp/aggregators-module/latest/_images/aggregator-size-timeout-diagram.png) Figure 1. サイズベースのアグリゲーターのタイムアウトと最大サイズの動作の図 - **\[Max size (最大サイズ)\]**​ 項目は、集約が完了とみなされる前に集約される要素の合計数を示します。 - **\[Timeout (タイムアウト)\]**​ 項目は集約が完了するのを待機する最大時間を示します。 - 最初のプロセスは、タイムアウトの発生前に集約された要素の合計数が必要な数に達したために集約が完了したことを示しています。 - 2 番目のプロセスは、タイムアウトの発生前に集約された要素の合計数が必要な数に達しなかったために集約が完了していないことを示しています。これにより、集約はリセットされます。 - 3 番目のプロセスは、​**\[Aggregator listener (アグリゲーターリスナー)\]**​ ソースを使用して、タイムアウトの発生前に集約された要素を取得できることを示しています。このソースは、タイムアウトが発生するとトリガーされます。 **\[Aggregator listener (アグリゲーターリスナー)\]**​ ソースを使用して未完了の集約要素を取得する場合、​**\[Include timed out group (タイムアウトグループを含める)\]**​ 項目を選択して、タイムアウトの発生時にソースがトリガーされることを示す必要があります。 ## [](#mule-アプリケーションの設定)Mule アプリケーションの設定 次の例は、​`10`​ 秒間に最大 ​`3`​ 件の HTTP 要求コールを集約するように ​**\[Size based Aggregator (サイズベースのアグリゲーター)\]**​ スコープを設定する方法を示しています。アプリケーションは DataWeave を使用して、ランダム ID 番号で各コールのペイロードを設定します。タイムアウトの発生前に集約の完了に必要な ​`3`​ 件のコールに達すると、アプリケーションはすべての集約された要素を記録します。それ以外の場合、アプリケーションは、タイムアウトの発生時に ​**\[Aggregator listener (アグリゲーターリスナー)\]**​ ソースによって取得された集約済み要素のみを記録します。 ![Anypoint Studio キャンバスでのサイズベースのアグリゲーターのフローの例](/mulesoft/current/ja-jp/aggregators-module/latest/_images/aggregator-size-timeout-example.png) Figure 2. サイズベースのアグリゲーターのフローの例 独自の環境でこの例をテストするには、Mule アプリケーションを作成し、curl コマンドを使用してアプリケーションを実行する必要があります。 Anypoint Studio で Mule フローを作成する手順は、次のとおりです。 1. **\[Mule Palette (Mule パレット)\]**​ で、​**\[HTTP Listener (HTTP リスナー)\]**​ を選択してキャンバスにドラッグします。 このソースは受信 HTTP メッセージ属性をリスンすることでフローを開始します。 2. **\[Connector configuration (コネクタ設定)\]**​ で ​**\[HTTP\_Listener\_config\]**​ を選択します。 3. **\[Path (パス)\]**​ を ​`/test`​ に設定します。 4. **\[Transform message\]**​ コンポーネントを ​**\[Listener\]**​ の右にドラッグします。 5. **\[Transform message\]**​ コンポーネントの DataSense プレビューウィンドウで、次の DataWeave 式を追加します。これにより、各 HTTP 要求のペイロードに ​`Source Name`​ が ​`"size-based"`​ として設定され、​`Source ID`​ が ​`random() as String`​ として設定されます。 ```DataWeave %dw 2.0 output application/json --- { "Source Name": "size-based", "Source ID": (random() as String) } ``` 6. **\[Size based aggregator (サイズベースのアグリゲーター)\]**​ スコープを ​**\[Transform message\]**​ の右にドラッグします。 7. **\[名前\]**​ を ​`sizeBasedAggregator`​ に設定します。 8. **\[Max Size (最大サイズ)\]**​ を ​`3`​ に設定します。 9. **\[Timeout (タイムアウト)\]**​ を ​`10`​ に設定します。 10. **\[Content (コンテンツ)\]**​ を ​`payload`​ に設定します。これは、何を集約するか (この場合は変換された HTTP 要求) を定義する式です。 11. **\[Logger\]**​ コンポーネントを ​**\[Incremental aggregation (増分集約)\]**​ ルートにドラッグします。 12. **\[Logger\]**​ 設定画面で、​**\[Message (メッセージ)\]**​ を ​`Doing incremental aggregation step.`​ に設定します。 13. **\[Logger\]**​ コンポーネントを ​**\[Aggregation complete (集約完了)\]**​ ルートにドラッグします。 14. **\[Message (メッセージ)\]**​ を ​`Aggregation complete`​ に設定します。 15. 別の ​**\[Logger\]**​ コンポーネントを ​**\[Aggregation complete (集約完了)\]**​ 内の最初の ​**\[Logger\]**​ コンポーネントの右にドラッグします。 16. **\[Message (メッセージ)\]**​ を次の式に設定します。この式では、すべての集約された要素が返されます。 ```xml output application/json --- payload ``` 17. **\[Aggregator listener (アグリゲーターリスナー)\]**​ ソースを最初のフローの下にドラッグします。 18. **\[Aggregator name (アグリゲーター名)\]**​ を ​`sizeBasedAggregator`​ に設定します。これは、前のアグリゲーターをリスンします。 19. **\[SelectInclude Timed Out Groups (タイムアウトグループを含める)\]**​ を選択します。これは、タイムアウトの発生時にリスナーがトリガーされるかどうかを示します。 20. **\[Logger\]**​ コンポーネントを ​**\[Aggregation listener (集約リスナー)\]**​ ソースの右にドラッグします。 21. **\[Message (メッセージ)\]**​ を次の式に設定します。この式では、タイムアウトの発生前に集約された要素のみが返されます。 ```xml output application/json --- payload ``` 22. プロジェクトを保存します。 23. **Package Explorer**​ で、​**\[Run (実行)\] > \[Run As (別のユーザーとして実行)\] > \[Mule Application (Mule アプリケーション)\]**​ をクリックします。 24. 10 秒以内に次の curl コマンドを送信して、アプリケーションをテストします。 `curl -X POST [http://localhost:8086/test](http://localhost:8086/test)` 25. **\[Console (コンソール)\]**​ ビューに移動して、ロガーメッセージを参照します。 ```none INFO 2021-06-22 12:59:16,319 [[MuleRuntime].uber.12: [aggregator-size-demo].aggregator-size-demoFlow.CPU_INTENSIVE @7992f689] [processor: aggregator-size-demoFlow/processors/1/route/0/processors/0; event: cbc219c0-d372-11eb-b975-f01898ad2638] org.mule.runtime.core.internal.processor.LoggerMessageProcessor: Doing incremental aggregation step. INFO 2021-06-22 12:59:26,328 [[MuleRuntime].uber.12: [aggregator-size-demo].aggregator-size-demoFlow1.CPU_LITE @684a1bef] [processor: aggregator-size-demoFlow1/processors/0; event: d6b48f00-d36f-11eb-b975-f01898ad2638] org.mule.runtime.core.internal.processor.LoggerMessageProcessor: [ { "Source Name": "size-based", "Source ID": "0.6568747259174192" } ] ``` このロガーメッセージの例は、最初の HTTP 要求で、​**\[Incremental aggregator (増分アグリゲーター)\]**​ ルートの ​**\[Logger\]**​ コンポーネントによってメッセージ ​`Doing incremental aggregation step.`​ が記録されることを示しています。集約が ​`10`​ 秒に達すると、タイムアウトが発生します。​**\[Aggregation listener (集約リスナー)\]**​ ソースによって唯一の HTTP 要求コールが取得され、後続の ​**\[Logger\]**​ コンポーネントによって、集約されたペイロード要素 (ランダム ID 番号) が返されます。 ## [](#サイズベースのアグリゲーターのタイムアウトと最大サイズの例の-xml)サイズベースのアグリゲーターのタイムアウトと最大サイズの例の XML この例のフローをすばやく Mule アプリケーションに読み込むには、次のコードを Studio XML エディターに貼り付けます。 ```xml ``` ## [](#関連情報)関連情報 - [Aggregators Module の例](https://docs.mulesoft.com/jp/aggregators-module/latest/aggregators-examples) - [MuleSoft ヘルプセンター](https://help.mulesoft.com) # サイズベースのアグリゲーターの例 Aggregators Module の ​**\[Size based aggregator (サイズベースのアグリゲーター)\]**​ スコープを使用すると、定義済みのサイズ数の要素の集約が完了するまで要素を集約することができます。次の例は、最大 ​`3`​ 件の HTTP 要求コールを集約するようにこのスコープを設定する方法を示しています。アプリケーションは DataWeave を使用して、ランダム ID 番号で各コールのペイロードを設定します。集約の完了に必要な ​`3`​ 件のコールに達すると、アプリケーションはすべての集約されたランダム ID 番号要素を記録します。 ![Anypoint Studio キャンバスでのサイズベースのアグリゲーターのフローの例](/mulesoft/current/ja-jp/aggregators-module/1.0/_images/aggregator-size-example.png) Figure 1. サイズベースのアグリゲーターのフローの例 独自の環境でこの例をテストするには、Mule アプリケーションを作成し、curl コマンドを使用してアプリケーションを実行する必要があります。 ## [](#mule-アプリケーションの作成)Mule アプリケーションの作成 Anypoint Studio で Mule フローを作成する手順は、次のとおりです。 1. **\[Mule Palette (Mule パレット)\]**​ で、​**\[HTTP Listener (HTTP リスナー)\]**​ を選択してキャンバスにドラッグします。 このソースは受信 HTTP メッセージ属性をリスンすることでフローを開始します。 2. **\[Connector configuration (コネクタ設定)\]**​ で ​**\[HTTP\_Listener\_config\]**​ を選択します。 3. **\[Path (パス)\]**​ を ​`/test`​ に設定します。 4. **\[Transform message\]**​ コンポーネントを ​**\[Listener\]**​ の右にドラッグします。 5. **\[Transform message\]**​ コンポーネントの DataSense プレビューウィンドウで、次の DataWeave 式を追加します。これにより、各 HTTP 要求のペイロードに ​`Source Name`​ が ​`"size-based"`​ として設定され、​`Source ID`​ が ​`random() as String`​ として設定されます。 ```DataWeave %dw 2.0 output application/json --- { "Source Name": "size-based", "Source ID": (random() as String) } ``` 6. **\[Size based aggregator (サイズベースのアグリゲーター)\]**​ スコープを ​**\[Transform message\]**​ の右にドラッグします。 7. **\[名前\]**​ を ​`sizeBasedAggregator`​ に設定します。 8. **\[Max Size (最大サイズ)\]**​ を ​`3`​ に設定します。 9. **\[Content (コンテンツ)\]**​ を ​`payload`​ に設定します。これは、何を集約するか (この場合は変換された HTTP 要求) を定義する式です。 10. **\[Logger\]**​ コンポーネントを ​**\[Incremental aggregation (増分集約)\]**​ ルートにドラッグします。 11. **\[Logger\]**​ 設定画面で、​**\[Message (メッセージ)\]**​ を ​`Doing incremental aggregation step.`​ に設定します。 12. **\[Logger\]**​ コンポーネントを ​**\[Aggregation complete (集約完了)\]**​ ルートにドラッグします。 13. **\[Message (メッセージ)\]**​ を ​`Aggregation complete`​ に設定します。 14. 別の ​**\[Logger\]**​ コンポーネントを ​**\[Aggregation complete (集約完了)\]**​ 内の最初の ​**\[Logger\]**​ コンポーネントの右にドラッグします。 15. **\[Message (メッセージ)\]**​ を次の式に設定します。この式では、すべての集約された要素が返されます。 ```xml output application/json --- payload ``` 16. プロジェクトを保存します。 17. **Package Explorer**​ で、​**\[Run (実行)\] > \[Run As (別のユーザーとして実行)\] > \[Mule Application (Mule アプリケーション)\]**​ をクリックします。 18. 次の curl コマンドを 3 回送信して、アプリケーションをテストします。 `curl -X POST [http://localhost:8086/test](http://localhost:8086/test)` これにより、​**\[Size based aggregator (サイズベースのアグリゲーター)\]**​ で処理する 3 個の HTTP 要求が生成されます。 19. **\[Console (コンソール)\]**​ ビューに移動して、ロガーメッセージを参照します。 ```none INFO 2021-06-08 11:18:44,897 [[MuleRuntime].uber.02: [aggregator-size-demo].aggregator-size-demoFlow.CPU_INTENSIVE @20a8a5b4] [processor: aggregator-size-demoFlow/processors/1/route/0/processors/0; event: 6ee63ec0-c864-11eb-b5a7-f01898ad2638] org.mule.runtime.core.internal.processor.LoggerMessageProcessor: Doing incremental aggregation step. INFO 2021-06-08 11:18:48,570 [[MuleRuntime].uber.02: [aggregator-size-demo].aggregator-size-demoFlow.CPU_INTENSIVE @20a8a5b4] [processor: aggregator-size-demoFlow/processors/1/route/0/processors/0; event: 71283f80-c864-11eb-b5a7-f01898ad2638] org.mule.runtime.core.internal.processor.LoggerMessageProcessor: Doing incremental aggregation step. INFO 2021-06-08 11:18:50,189 [[MuleRuntime].uber.02: [aggregator-size-demo].aggregator-size-demoFlow.CPU_INTENSIVE @20a8a5b4] [processor: aggregator-size-demoFlow/processors/1/route/1/processors/0; event: 721efb90-c864-11eb-b5a7-f01898ad2638] org.mule.runtime.core.internal.processor.LoggerMessageProcessor: Aggregation complete INFO 2021-06-08 11:18:50,254 [[MuleRuntime].uber.02: [aggregator-size-demo].aggregator-size-demoFlow.CPU_INTENSIVE @20a8a5b4] [processor: aggregator-size-demoFlow/processors/1/route/1/processors/1; event: 721efb90-c864-11eb-b5a7-f01898ad2638] org.mule.runtime.core.internal.processor.LoggerMessageProcessor: [ { "Source Name": "size-based", "Source ID": "0.49127866969631506" }, { "Source Name": "size-based", "Source ID": "0.4250959269275998" }, { "Source Name": "size-based", "Source ID": "0.4160926509813687" } ] ``` このロガーメッセージの例は、最初の 2 つの HTTP 要求で、​**\[Incremental aggregator (増分アグリゲーター)\]**​ ルートの ​**\[Logger\]**​ コンポーネントによってメッセージ ​`Doing incremental aggregation step.`​ が記録されることを示しています。集約が 3 番目の HTTP 要求コールに達すると、​**\[Aggregation complete (集約完了)\]**​ ルートの最初の ​**\[Logger\]**​ コンポーネントによってメッセージ ​`Aggregation complete`​ が記録され、2 番目の ​**\[Logger\]**​ コンポーネントによって、集約されたペイロード要素 (ランダム ID 番号) が返されます。 ```none { "Source Name": "size-based", "Source ID": "0.49127866969631506" }, { "Source Name": "size-based", "Source ID": "0.4250959269275998" }, { "Source Name": "size-based", "Source ID": "0.4160926509813687" } ``` ## [](#サイズベースのアグリゲーターの例の-xml)サイズベースのアグリゲーターの例の XML この例のフローをすばやく Mule アプリケーションに読み込むには、次のコードを Studio XML エディターに貼り付けます。 ```xml ``` ## [](#関連情報)関連情報 - [Aggregators Module の例](https://docs.mulesoft.com/jp/aggregators-module/1.0/aggregators-examples) - [MuleSoft ヘルプセンター](https://help.mulesoft.com) # サイズベースのアグリゲーターの例 Aggregators Module の ​**\[Size based aggregator (サイズベースのアグリゲーター)\]**​ スコープを使用すると、定義済みのサイズ数の要素の集約が完了するまで要素を集約することができます。次の例は、最大 ​`3`​ 件の HTTP 要求コールを集約するようにこのスコープを設定する方法を示しています。アプリケーションは DataWeave を使用して、ランダム ID 番号で各コールのペイロードを設定します。集約の完了に必要な ​`3`​ 件のコールに達すると、アプリケーションはすべての集約されたランダム ID 番号要素を記録します。 ![Anypoint Studio キャンバスでのサイズベースのアグリゲーターのフローの例](/mulesoft/current/ja-jp/aggregators-module/latest/_images/aggregator-size-example.png) Figure 1. サイズベースのアグリゲーターのフローの例 独自の環境でこの例をテストするには、Mule アプリケーションを作成し、curl コマンドを使用してアプリケーションを実行する必要があります。 ## [](#mule-アプリケーションの作成)Mule アプリケーションの作成 Anypoint Studio で Mule フローを作成する手順は、次のとおりです。 1. **\[Mule Palette (Mule パレット)\]**​ で、​**\[HTTP Listener (HTTP リスナー)\]**​ を選択してキャンバスにドラッグします。 このソースは受信 HTTP メッセージ属性をリスンすることでフローを開始します。 2. **\[Connector configuration (コネクタ設定)\]**​ で ​**\[HTTP\_Listener\_config\]**​ を選択します。 3. **\[Path (パス)\]**​ を ​`/test`​ に設定します。 4. **\[Transform message\]**​ コンポーネントを ​**\[Listener\]**​ の右にドラッグします。 5. **\[Transform message\]**​ コンポーネントの DataSense プレビューウィンドウで、次の DataWeave 式を追加します。これにより、各 HTTP 要求のペイロードに ​`Source Name`​ が ​`"size-based"`​ として設定され、​`Source ID`​ が ​`random() as String`​ として設定されます。 ```DataWeave %dw 2.0 output application/json --- { "Source Name": "size-based", "Source ID": (random() as String) } ``` 6. **\[Size based aggregator (サイズベースのアグリゲーター)\]**​ スコープを ​**\[Transform message\]**​ の右にドラッグします。 7. **\[名前\]**​ を ​`sizeBasedAggregator`​ に設定します。 8. **\[Max Size (最大サイズ)\]**​ を ​`3`​ に設定します。 9. **\[Content (コンテンツ)\]**​ を ​`payload`​ に設定します。これは、何を集約するか (この場合は変換された HTTP 要求) を定義する式です。 10. **\[Logger\]**​ コンポーネントを ​**\[Incremental aggregation (増分集約)\]**​ ルートにドラッグします。 11. **\[Logger\]**​ 設定画面で、​**\[Message (メッセージ)\]**​ を ​`[Doing incremental aggregation step]`​ に設定します。 12. **\[Logger\]**​ コンポーネントを ​**\[Aggregation complete (集約完了)\]**​ ルートにドラッグします。 13. **\[Message (メッセージ)\]**​ を ​`Aggregation complete`​ に設定します。 14. 別の ​**\[Logger\]**​ コンポーネントを ​**\[Aggregation complete (集約完了)\]**​ 内の最初の ​**\[Logger\]**​ コンポーネントの右にドラッグします。 15. **\[Message (メッセージ)\]**​ を次の式に設定します。この式では、すべての集約された要素が返されます。 ```xml output application/json --- payload ``` 16. プロジェクトを保存します。 17. **Package Explorer**​ で、​**\[Run (実行)\] > \[Run As (別のユーザーとして実行)\] > \[Mule Application (Mule アプリケーション)\]**​ をクリックします。 18. 次の curl コマンドを 3 回送信して、アプリケーションをテストします。 `curl -X POST [http://localhost:8086/test](http://localhost:8086/test)` これにより、​**\[Size based aggregator (サイズベースのアグリゲーター)\]**​ で処理する 3 個の HTTP 要求が生成されます。 19. **\[Console (コンソール)\]**​ ビューに移動して、ロガーメッセージを参照します。 ```none INFO 2021-06-08 11:18:44,897 [[MuleRuntime].uber.02: [aggregator-size-demo].aggregator-size-demoFlow.CPU_INTENSIVE @20a8a5b4] [processor: aggregator-size-demoFlow/processors/1/route/0/processors/0; event: 6ee63ec0-c864-11eb-b5a7-f01898ad2638] org.mule.runtime.core.internal.processor.LoggerMessageProcessor: Doing incremental aggregation step. INFO 2021-06-08 11:18:48,570 [[MuleRuntime].uber.02: [aggregator-size-demo].aggregator-size-demoFlow.CPU_INTENSIVE @20a8a5b4] [processor: aggregator-size-demoFlow/processors/1/route/0/processors/0; event: 71283f80-c864-11eb-b5a7-f01898ad2638] org.mule.runtime.core.internal.processor.LoggerMessageProcessor: Doing incremental aggregation step. INFO 2021-06-08 11:18:50,189 [[MuleRuntime].uber.02: [aggregator-size-demo].aggregator-size-demoFlow.CPU_INTENSIVE @20a8a5b4] [processor: aggregator-size-demoFlow/processors/1/route/1/processors/0; event: 721efb90-c864-11eb-b5a7-f01898ad2638] org.mule.runtime.core.internal.processor.LoggerMessageProcessor: Aggregation complete INFO 2021-06-08 11:18:50,254 [[MuleRuntime].uber.02: [aggregator-size-demo].aggregator-size-demoFlow.CPU_INTENSIVE @20a8a5b4] [processor: aggregator-size-demoFlow/processors/1/route/1/processors/1; event: 721efb90-c864-11eb-b5a7-f01898ad2638] org.mule.runtime.core.internal.processor.LoggerMessageProcessor: [ { "Source Name": "size-based", "Source ID": "0.49127866969631506" }, { "Source Name": "size-based", "Source ID": "0.4250959269275998" }, { "Source Name": "size-based", "Source ID": "0.4160926509813687" } ] ``` このロガーメッセージの例は、最初の 2 つの HTTP 要求で、​**\[Incremental aggregator (増分アグリゲーター)\]**​ ルートの ​**\[Logger\]**​ コンポーネントによってメッセージ ​`Doing incremental aggregation step.`​ が記録されることを示しています。集約が 3 番目の HTTP 要求コールに達すると、​**\[Aggregation complete (集約完了)\]**​ ルートの最初の ​**\[Logger\]**​ コンポーネントによってメッセージ ​`Aggregation complete`​ が記録され、2 番目の ​**\[Logger\]**​ コンポーネントによって、集約されたペイロード要素 (ランダム ID 番号) が返されます。 ```none { "Source Name": "size-based", "Source ID": "0.49127866969631506" }, { "Source Name": "size-based", "Source ID": "0.4250959269275998" }, { "Source Name": "size-based", "Source ID": "0.4160926509813687" } ``` ## [](#サイズベースのアグリゲーターの例の-xml)サイズベースのアグリゲーターの例の XML この例のフローをすばやく Mule アプリケーションに読み込むには、次のコードを Studio XML エディターに貼り付けます。 ```xml ``` ## [](#関連情報)関連情報 - [Aggregators Module の例](https://docs.mulesoft.com/jp/aggregators-module/latest/aggregators-examples) - [MuleSoft ヘルプセンター](https://help.mulesoft.com) # 同期集約と非同期集約の比較 集約は、リストに追加される新しい要素に基づいて完了するか (最大サイズが指定されている場合)、タイムアウトや期間の完了によって完了します。要素のリストを使用して実行するコンポーネントのチェーンは集約の種別によって決まるため、これは重要です。 集約完了のトリガーには、同期と非同期の 2 つがあります。
アグリゲーターに関連付けられた各時間カウンターは、グループの最初のメッセージが到着した瞬間から計数を開始します。集約が完了すると、カウンターはリセットされ、次の要素の到着を待ちます。
単一グループのアグリゲーター (時間ベースおよびサイズベースのアグリゲータースコープ) の場合、時間カウンターは 1 つのみですが、グループベースのアグリゲーターの場合、グループごとに 1 つのカウンターがあります。
## [](#同期集約)同期集約 新しい要素が到着したことで完了した集約の場合 (同期)、次のうち少なくとも 1 つの動作が発生します。 - アグリゲータースコープが ​**\[Aggregation complete (集約完了)\]**​ ルートで設定されている場合、そのルート内のコンポーネントは、集約された要素のリストであるペイロードを使用して実行されます。 - アグリゲータースコープに、参照されている ​**\[Aggregator listener (アグリゲーターリスナー)\]**​ がある場合、そのリスナーが属するフローは、集約された要素のリストであるペイロードを使用して実行されます。 ## [](#非同期集約)非同期集約 期間が終了したまたはタイムアウトに達したことで完了した集約の場合 (非同期)、次の 1 つの動作が発生する可能性があります。アグリゲータースコープに、参照されている ​**\[Aggregator listener (アグリゲーターリスナー)\]**​ があり、リスナーでタイムアウト集約を受け入れている場合、そのリスナーが属するフローは、集約された要素のリストをペイロードとして実行します。 ## [](#アグリゲーターを使用する-mule-アプリケーションの処理ロジックの配置)アグリゲーターを使用する Mule アプリケーションの処理ロジックの配置 アグリゲーターを使用する Mule アプリケーションを作成する場合、処理ロジックを次のように配置します。 - 集約が時間に依存する場合、​**\[Aggregation listener (集約リスナー)\]**​ をソースとする異なるフローに処理ロジックを追加します。 - 到達したサイズに依存する場合、​**\[Aggregation complete (集約完了)\]**​ ルートに処理ロジックを追加します。 - 時間とサイズの両方に依存する場合は、次のように配置します。 - ソースのないサブフローにメインロジックを追加します。​**\[Is aggregation complete (集約完了)\]**​ 属性で、集約が終了した方法が示されます。 - **\[Aggregation complete (集約完了)\]**​ ルート内のメインロジックフローに ​**Flow reference**​ コンポーネントを追加します。サブフローがこのフロー参照により実行される場合、​**\[Is aggregation complete (集約完了)\]**​ 属性は ​`true`​ になります。 - アグリゲーターをリスンし、タイムアウトグループを受け入れる ​**\[Aggregation listener (集約リスナー)\]**​ を使用する別のフローを追加します。リスナーの後に、メインロジックサブフローをコールする ​**Flow reference**​ コンポーネントを続けます。この場合、​**\[Is aggregation complete (集約完了)\]**​ 属性は ​`false`​ になります。 ## [](#関連情報)関連情報 - [Aggregators Module リファレンス](https://docs.mulesoft.com/jp/aggregators-module/1.0/aggregators-module-reference) - [MuleSoft ヘルプセンター](https://help.mulesoft.com) # 同期集約と非同期集約の比較 集約は、リストに追加される新しい要素に基づいて完了するか (最大サイズが指定されている場合)、タイムアウトや期間の完了によって完了します。要素のリストを使用して実行するコンポーネントのチェーンは集約の種別によって決まるため、これは重要です。 集約完了のトリガーには、同期と非同期の 2 つがあります。
アグリゲーターに関連付けられた各時間カウンターは、グループの最初のメッセージが到着した瞬間から計数を開始します。集約が完了すると、カウンターはリセットされ、次の要素の到着を待ちます。
単一グループのアグリゲーター (時間ベースおよびサイズベースのアグリゲータースコープ) の場合、時間カウンターは 1 つのみですが、グループベースのアグリゲーターの場合、グループごとに 1 つのカウンターがあります。
## [](#同期集約)同期集約 新しい要素が到着したことで完了した集約の場合 (同期)、次のうち少なくとも 1 つの動作が発生します。 - アグリゲータースコープが ​**\[Aggregation complete (集約完了)\]**​ ルートで設定されている場合、そのルート内のコンポーネントは、集約された要素のリストであるペイロードを使用して実行されます。 - アグリゲータースコープに、参照されている ​**\[Aggregator listener (アグリゲーターリスナー)\]**​ がある場合、そのリスナーが属するフローは、集約された要素のリストであるペイロードを使用して実行されます。 ## [](#非同期集約)非同期集約 期間が終了したまたはタイムアウトに達したことで完了した集約の場合 (非同期)、次の 1 つの動作が発生する可能性があります。アグリゲータースコープに、参照されている ​**\[Aggregator listener (アグリゲーターリスナー)\]**​ があり、リスナーでタイムアウト集約を受け入れている場合、そのリスナーが属するフローは、集約された要素のリストをペイロードとして実行します。 ## [](#アグリゲーターを使用する-mule-アプリケーションの処理ロジックの配置)アグリゲーターを使用する Mule アプリケーションの処理ロジックの配置 アグリゲーターを使用する Mule アプリケーションを作成する場合、処理ロジックを次のように配置します。 - 集約が時間に依存する場合、​**\[Aggregation listener (集約リスナー)\]**​ をソースとする異なるフローに処理ロジックを追加します。 - 到達したサイズに依存する場合、​**\[Aggregation complete (集約完了)\]**​ ルートに処理ロジックを追加します。 - 時間とサイズの両方に依存する場合は、次のように配置します。 - ソースのないサブフローにメインロジックを追加します。​**\[Is aggregation complete (集約完了)\]**​ 属性で、集約が終了した方法が示されます。 - **\[Aggregation complete (集約完了)\]**​ ルート内のメインロジックフローに ​**Flow reference**​ コンポーネントを追加します。サブフローがこのフロー参照により実行される場合、​**\[Is aggregation complete (集約完了)\]**​ 属性は ​`true`​ になります。 - アグリゲーターをリスンし、タイムアウトグループを受け入れる ​**\[Aggregation listener (集約リスナー)\]**​ を使用する別のフローを追加します。リスナーの後に、メインロジックサブフローをコールする ​**Flow reference**​ コンポーネントを続けます。この場合、​**\[Is aggregation complete (集約完了)\]**​ 属性は ​`false`​ になります。 ## [](#関連情報)関連情報 - [Aggregators Module リファレンス](https://docs.mulesoft.com/jp/aggregators-module/latest/aggregators-module-reference) - [MuleSoft ヘルプセンター](https://help.mulesoft.com) # 時間ベースのアグリゲーターの例 Aggregators Module の ​**\[Time based aggregator (時間ベースのアグリゲーター)\]**​ スコープを使用すると、定義済みの時間制限内で要素を集約することができます。次の例は、​`5`​ 秒間の HTTP 要求コールを集約するようにこのスコープを設定する方法を示しています。アプリケーションは DataWeave を使用して、ランダム ID 番号で各コールのペイロードを設定します。集約が ​`5`​ 秒に達すると、​**\[Aggregator listener (アグリゲーターリスナー)\]**​ ソースは要素のリストを取得し、アプリケーションはすべての集約されたランダム ID 番号を記録します。 ![Anypoint Studio キャンバスでの時間ベースのアグリゲーターのフローの例](/mulesoft/current/ja-jp/aggregators-module/1.0/_images/aggregator-time-example.png) Figure 1. 時間ベースのアグリゲーターのフローの例 独自の環境でこの例をテストするには、Mule アプリケーションを作成し、curl コマンドを使用してアプリケーションを実行する必要があります。 ## [](#mule-アプリケーションの作成)Mule アプリケーションの作成 Anypoint Studio で Mule フローを作成する手順は、次のとおりです。 1. **\[Mule Palette (Mule パレット)\]**​ で、​**\[HTTP Listener (HTTP リスナー)\]**​ を選択してキャンバスにドラッグします。 このソースは受信 HTTP メッセージ属性をリスンすることでフローを開始します。 2. **\[Connector configuration (コネクタ設定)\]**​ で ​**\[HTTP\_Listener\_config\]**​ を選択します。 3. **\[Path (パス)\]**​ を ​`/test`​ に設定します。 4. **\[Transform message\]**​ コンポーネントを ​**\[Listener\]**​ の右にドラッグします。 5. **\[Transform message\]**​ コンポーネントの DataSense プレビューウィンドウで、次の DataWeave 式を追加します。これにより、各 HTTP 要求のペイロードに ​`Source Name`​ が ​`"time-based"`​ として設定され、​`Source ID`​ が ​`random() as String`​ として設定されます。 ```DataWeave %dw 2.0 output application/json --- { "Source Name": "time-based", "Source ID": (random() as String) } ``` 6. **\[Time based aggregator (時間ベースのアグリゲーター)\]**​ スコープを ​**\[Transform message\]**​ の右にドラッグします。 7. **\[名前\]**​ を ​`timeBasedAggregator`​ に設定します。 8. **\[Period (期間)\]**​ を ​`5`​ に設定します。 9. **\[Period unit (期間単位)\]**​ を ​`SECONDS(Default)`​ に設定します。 10. **\[Content (コンテンツ)\]**​ を ​`payload`​ に設定します。これは、何を集約するか (この場合は変換された HTTP 要求) を定義する式です。 11. **\[Logger\]**​ コンポーネントを ​**\[Incremental aggregation (増分集約)\]**​ ルートにドラッグします。 12. **\[Logger\]**​ 設定画面で、​**\[Message (メッセージ)\]**​ を ​`Doing incremental step.`​ に設定します。 13. **\[Aggregator listener (アグリゲーターリスナー)\]**​ ソースを最初のフローの下にドラッグします。 このソースは、前に定義された時間ベースのアグリゲーターをリスンします。このアグリゲーターがその要素を解放すると、リスナーが実行されます。 14. **\[Aggregator name (アグリゲーター名)\]**​ を ​`timeBasedAggregator`​ に設定します。 15. **\[Logger\]**​ コンポーネントを ​**\[Aggregator listener (アグリゲーターリスナー)\]**​ の右にドラッグします 16. **\[Message (メッセージ)\]**​ を ​`Aggregation complete`​ に設定します。 17. 別の ​**\[Logger\]**​ コンポーネントを、前に追加した ​**\[Logger\]**​ コンポーネントの右にドラッグします。 18. **\[Message (メッセージ)\]**​ を次の式に設定します。この式では、すべての集約された要素が返されます。 ```xml output application/json --- payload ``` 19. プロジェクトを保存します。 20. **Package Explorer**​ で、​**\[Run (実行)\] > \[Run As (別のユーザーとして実行)\] > \[Mule Application (Mule アプリケーション)\]**​ をクリックします。 21. `5`​ 秒以内に次の curl コマンドを 2 回送信して、アプリケーションをテストします。 `curl -X POST [http://localhost:8086/test](http://localhost:8086/test)`​ これにより、​**\[Time based aggregator (時間ベースのアグリゲーター)\]**​ で処理する 2 個の HTTP 要求が生成されます。 22. **\[Console (コンソール)\]**​ ビューに移動して、ロガーメッセージを参照します。 ```none INFO 2021-06-09 12:02:04,203 [[MuleRuntime].uber.04: [aggregator-time-demo].aggregator-time-demoFlow.CPU_INTENSIVE @59b15997] [processor: aggregator-time-demoFlow/processors/1/route/0/processors/0; event: a6b00280-c933-11eb-a825-f01898ad2638] org.mule.runtime.core.internal.processor.LoggerMessageProcessor: Doing incremental step. INFO 2021-06-09 12:02:06,071 [[MuleRuntime].uber.04: [aggregator-time-demo].aggregator-time-demoFlow.CPU_INTENSIVE @59b15997] [processor: aggregator-time-demoFlow/processors/1/route/0/processors/0; event: a7cd0b40-c933-11eb-a825-f01898ad2638] org.mule.runtime.core.internal.processor.LoggerMessageProcessor: Doing incremental step. INFO 2021-06-09 12:02:09,206 [[MuleRuntime].uber.04: [aggregator-time-demo].aggregator-time-demoOnCompleteFlow.CPU_LITE @2ab3c271] [processor: aggregator-time-demoOnCompleteFlow/processors/0; event: 827c9591-c933-11eb-a825-f01898ad2638] org.mule.runtime.core.internal.processor.LoggerMessageProcessor: Aggregation complete INFO 2021-06-09 12:02:09,208 [[MuleRuntime].uber.04: [aggregator-time-demo].aggregator-time-demoOnCompleteFlow.CPU_LITE @2ab3c271] [processor: aggregator-time-demoOnCompleteFlow/processors/1; event: 827c9591-c933-11eb-a825-f01898ad2638] org.mule.runtime.core.internal.processor.LoggerMessageProcessor: [ { "Source Name": "time-based", "Source ID": "0.7095971146328517" }, { "Source Name": "time-based", "Source ID": "0.6900116288202388" } ] ``` このロガーメッセージの例は、各 HTTP 要求で、​**\[Incremental aggregator (増分アグリゲーター)\]**​ ルートの ​**\[Logger\]**​ コンポーネントによってメッセージ ​`Doing incremental step`​ が記録されることを示しています。​**\[Aggregator listener (アグリゲーターリスナー)\]**​ ソースによって参照される ​`timeBasedAggregator`​ 値が ​`5`​ 秒以内に増分集計を完了すると、すべての要素のリストを使用してリスナーがトリガーされます。後続の ​**\[Logger\]**​ コンポーネントによってメッセージ ​`Aggregation complete`​ が記録され、2 番目の ​**\[Logger\]**​ コンポーネントによって、集約されたペイロード要素 (ランダム ID 番号) が返されます。 ```none { "Source Name": "time-based", "Source ID": "0.7095971146328517" }, { "Source Name": "time-based", "Source ID": "0.6900116288202388" } ``` ## [](#時間ベースのアグリゲーターの例の-xml)時間ベースのアグリゲーターの例の XML この例のフローをすばやく Mule アプリケーションに読み込むには、次のコードを Studio XML エディターに貼り付けます。 ```xml ``` ## [](#関連情報)関連情報 - [Aggregators Module の例](https://docs.mulesoft.com/jp/aggregators-module/1.0/aggregators-examples) - [MuleSoft ヘルプセンター](https://help.mulesoft.com) # 時間ベースのアグリゲーターの例 Aggregators Module の ​**\[Time based aggregator (時間ベースのアグリゲーター)\]**​ スコープを使用すると、定義済みの時間制限内で要素を集約することができます。次の例は、​`5`​ 秒間の HTTP 要求コールを集約するようにこのスコープを設定する方法を示しています。アプリケーションは DataWeave を使用して、ランダム ID 番号で各コールのペイロードを設定します。集約が ​`5`​ 秒に達すると、​**\[Aggregator listener (アグリゲーターリスナー)\]**​ ソースは要素のリストを取得し、アプリケーションはすべての集約されたランダム ID 番号を記録します。 ![Anypoint Studio キャンバスでの時間ベースのアグリゲーターのフローの例](/mulesoft/current/ja-jp/aggregators-module/latest/_images/aggregator-time-example.png) Figure 1. 時間ベースのアグリゲーターのフローの例 独自の環境でこの例をテストするには、Mule アプリケーションを作成し、curl コマンドを使用してアプリケーションを実行する必要があります。 ## [](#mule-アプリケーションの作成)Mule アプリケーションの作成 Anypoint Studio で Mule フローを作成する手順は、次のとおりです。 1. **\[Mule Palette (Mule パレット)\]**​ で、​**\[HTTP Listener (HTTP リスナー)\]**​ を選択してキャンバスにドラッグします。 このソースは受信 HTTP メッセージ属性をリスンすることでフローを開始します。 2. **\[Connector configuration (コネクタ設定)\]**​ で ​**\[HTTP\_Listener\_config\]**​ を選択します。 3. **\[Path (パス)\]**​ を ​`/test`​ に設定します。 4. **\[Transform message\]**​ コンポーネントを ​**\[Listener\]**​ の右にドラッグします。 5. **\[Transform message\]**​ コンポーネントの DataSense プレビューウィンドウで、次の DataWeave 式を追加します。これにより、各 HTTP 要求のペイロードに ​`Source Name`​ が ​`"time-based"`​ として設定され、​`Source ID`​ が ​`random() as String`​ として設定されます。 ```DataWeave %dw 2.0 output application/json --- { "Source Name": "time-based", "Source ID": (random() as String) } ``` 6. **\[Time based aggregator (時間ベースのアグリゲーター)\]**​ スコープを ​**\[Transform message\]**​ の右にドラッグします。 7. **\[名前\]**​ を ​`timeBasedAggregator`​ に設定します。 8. **\[Period (期間)\]**​ を ​`5`​ に設定します。 9. **\[Period unit (期間単位)\]**​ を ​`SECONDS(Default)`​ に設定します。 10. **\[Content (コンテンツ)\]**​ を ​`payload`​ に設定します。これは、何を集約するか (この場合は変換された HTTP 要求) を定義する式です。 11. **\[Logger\]**​ コンポーネントを ​**\[Incremental aggregation (増分集約)\]**​ ルートにドラッグします。 12. **\[Logger\]**​ 設定画面で、​**\[Message (メッセージ)\]**​ を ​`Doing incremental step.`​ に設定します。 13. **\[Aggregator listener (アグリゲーターリスナー)\]**​ ソースを最初のフローの下にドラッグします。 このソースは、前に定義された時間ベースのアグリゲーターをリスンします。このアグリゲーターがその要素を解放すると、リスナーが実行されます。 14. **\[Aggregator name (アグリゲーター名)\]**​ を ​`timeBasedAggregator`​ に設定します。 15. **\[Logger\]**​ コンポーネントを ​**\[Aggregator listener (アグリゲーターリスナー)\]**​ の右にドラッグします 16. **\[Message (メッセージ)\]**​ を ​`Aggregation complete`​ に設定します。 17. 別の ​**\[Logger\]**​ コンポーネントを、前に追加した ​**\[Logger\]**​ コンポーネントの右にドラッグします。 18. **\[Message (メッセージ)\]**​ を次の式に設定します。この式では、すべての集約された要素が返されます。 ```xml output application/json --- payload ``` 19. プロジェクトを保存します。 20. **Package Explorer**​ で、​**\[Run (実行)\] > \[Run As (別のユーザーとして実行)\] > \[Mule Application (Mule アプリケーション)\]**​ をクリックします。 21. `5`​ 秒以内に次の curl コマンドを 2 回送信して、アプリケーションをテストします。 `curl -X POST [http://localhost:8086/test](http://localhost:8086/test)`​ これにより、​**\[Time based aggregator (時間ベースのアグリゲーター)\]**​ で処理する 2 個の HTTP 要求が生成されます。 22. **\[Console (コンソール)\]**​ ビューに移動して、ロガーメッセージを参照します。 ```none INFO 2021-06-09 12:02:04,203 [[MuleRuntime].uber.04: [aggregator-time-demo].aggregator-time-demoFlow.CPU_INTENSIVE @59b15997] [processor: aggregator-time-demoFlow/processors/1/route/0/processors/0; event: a6b00280-c933-11eb-a825-f01898ad2638] org.mule.runtime.core.internal.processor.LoggerMessageProcessor: Doing incremental step. INFO 2021-06-09 12:02:06,071 [[MuleRuntime].uber.04: [aggregator-time-demo].aggregator-time-demoFlow.CPU_INTENSIVE @59b15997] [processor: aggregator-time-demoFlow/processors/1/route/0/processors/0; event: a7cd0b40-c933-11eb-a825-f01898ad2638] org.mule.runtime.core.internal.processor.LoggerMessageProcessor: Doing incremental step. INFO 2021-06-09 12:02:09,206 [[MuleRuntime].uber.04: [aggregator-time-demo].aggregator-time-demoOnCompleteFlow.CPU_LITE @2ab3c271] [processor: aggregator-time-demoOnCompleteFlow/processors/0; event: 827c9591-c933-11eb-a825-f01898ad2638] org.mule.runtime.core.internal.processor.LoggerMessageProcessor: Aggregation complete INFO 2021-06-09 12:02:09,208 [[MuleRuntime].uber.04: [aggregator-time-demo].aggregator-time-demoOnCompleteFlow.CPU_LITE @2ab3c271] [processor: aggregator-time-demoOnCompleteFlow/processors/1; event: 827c9591-c933-11eb-a825-f01898ad2638] org.mule.runtime.core.internal.processor.LoggerMessageProcessor: [ { "Source Name": "time-based", "Source ID": "0.7095971146328517" }, { "Source Name": "time-based", "Source ID": "0.6900116288202388" } ] ``` このロガーメッセージの例は、各 HTTP 要求で、​**\[Incremental aggregator (増分アグリゲーター)\]**​ ルートの ​**\[Logger\]**​ コンポーネントによってメッセージ ​`Doing incremental step`​ が記録されることを示しています。​**\[Aggregator listener (アグリゲーターリスナー)\]**​ ソースによって参照される ​`timeBasedAggregator`​ 値が ​`5`​ 秒以内に増分集計を完了すると、すべての要素のリストを使用してリスナーがトリガーされます。後続の ​**\[Logger\]**​ コンポーネントによってメッセージ ​`Aggregation complete`​ が記録され、2 番目の ​**\[Logger\]**​ コンポーネントによって、集約されたペイロード要素 (ランダム ID 番号) が返されます。 ```none { "Source Name": "time-based", "Source ID": "0.7095971146328517" }, { "Source Name": "time-based", "Source ID": "0.6900116288202388" } ``` ## [](#時間ベースのアグリゲーターの例の-xml)時間ベースのアグリゲーターの例の XML この例のフローをすばやく Mule アプリケーションに読み込むには、次のコードを Studio XML エディターに貼り付けます。 ```xml ``` ## [](#関連情報)関連情報 - [Aggregators Module の例](https://docs.mulesoft.com/jp/aggregators-module/latest/aggregators-examples) - [MuleSoft ヘルプセンター](https://help.mulesoft.com)