集約ファンアウト・フローの作成

集約ファンアウト・フローは、初期入力メッセージを受け取ると、それを再構築してさまざまなターゲット・アプリケーションへの要求を表示します。

始める前に:

同じメッセージ・フロー内にファンアウトおよびファンイン・フローを組み込むことは可能です。ただし、2 つの別々のフローを作成してもかまいません。 別個のメッセージ・フローを構成することの利点について詳しくは、ファンアウトおよびファンイン集約フローの関連付けを参照してください。

WebSphere® Message Broker で提供されているファンアウト・フローの例について検討するには、以下のサンプルを参照してください。 サンプルは、Message Brokers Toolkit と統合されているインフォメーション・センターを使用する場合にのみ表示できます。

ファンアウト・フローを作成するには、以下のようにします。

  1. 「ブローカー・アプリケーション開発」パースペクティブに切り替えます。
  2. ファンアウト処理を提供するために新しいメッセージ・フローを作成します。
  3. 以下のノードをエディター・ビューに追加して、以下に示されているとおりにこれらを構成および接続します。
    Input ノード
    入力ノードは、多重要求メッセージが生成元である入力メッセージを受け取ります。 このノードは、組み込みノードのいずれかまたはユーザー定義の入力ノードにすることができます。
    1. 入力ノードを選択して「プロパティー」ビューをオープンするか、またはノードを右クリックして「プロパティー」をクリックし、「プロパティー」ダイアログを表示します。 ノードのプロパティーが表示されます。
    2. このノードの入力メッセージのソースを指定します。 例えば、MQInput ノードがメッセージの検索元とする基本プロパティー「キュー名」内に、 WebSphere MQ キューの名前を指定します。
    3. オプション: このノードのために構成したいその他のプロパティーの値を指定します。 たとえば、拡張プロパティーの「トランザクション・モード」をデフォルトの「はい」に設定し、集約要求メッセージが同期点の下に書き込まれるようにします。 このオプションにより、AggregateReply ノードが、集約インスタンスについて通知する制御メッセージを受け取る前に応答メッセージを受け取るという状況が回避されます。 ファンアウト・フローをトランザクション制御の下に置くことで、 応答メッセージが AggregateReply に到達する前にファンアウト・フローを完了させることができます。
    4. 入力ノードの Out ターミナルを、AggregateControl ノードの In ターミナルに接続します。 このオプションは、最も単純な構成を表します。適当であれば、入力ノードと AggregateControl ノードとの間に他のノードを組み込むことができます。 例えば、監査のために要求を (Warehouse ノードで) 保管したり、 または固有 ID をメッセージに (Compute ノードで) 追加したい場合もあります。
    5. オプション: ファンアウトおよびファンイン・フローが 1 つのメッセージ・フロー内で結合されている場合は、「拡張」タブの「順序モード」プロパティーを変更します。「キュー順序」プロパティーを選択し、「論理順序」オプションも選択されていることを確認します。これらのオプションによって、 キューに到着するメッセージの論理順序が維持されるよう、入力ノードが単一スレッドとなります。 使用可能にする追加のインスタンス・スレッドはすべて、ファンイン入力ノード間でのみ共用され、集約のパフォーマンスが改善されます。 ファンインおよびファンアウト・フローが別々のメッセージ・フローに存在する場合は、追加のスレッドを特にファンイン・フローに対して使用可能にすることができるため、このステップは不要です。
    AggregateControl ノード
    AggregateControl ノードは、AggregateRequest ノードが必要とする情報を持つ入力メッセージに関連した AggregateRequest を更新します。 AggregateControl ノードは、LocalEnvironment.ComIbmAggregateControlNode フォルダーを作成します。 このフォルダーとその中にあるフィールドは、WebSphere Message Broker が内部で使用するためのものであり、メッセージ・フローを開発する時にそれらのフィールドの存在やそれらの値に依存すべきではありません。
    1. AggregateControl ノードを選択して「プロパティー」ビューをオープンするか、またはノードを右クリックして「プロパティー」をクリックし、「プロパティー」ダイアログを表示します。 ノードのプロパティーが表示されます。
    2. この特定の集約を識別するには、AggregateControl ノードの「集約名」>プロパティーを設定します。このプロパティーは、後にこの AggregateControl ノードを、 特定の AggregateReply ノードに関連付けるために使用されます。 指定する「集約名」は、ブローカー内でコンテキストに応じて固有でなければなりません。
    3. オプション: 「タイムアウト」プロパティーを設定して、 応答がいくつかのアクションを実行するまでブローカーが待機する時間を指定します (集約のタイムアウトの設定で説明されています)。 AggregateControl でタイムアウトが設定されていない場合、最初に格納された集約要求は、 すべての集約要求応答メッセージが戻されるまで除去されません。 この状態により、内部キューにメッセージが徐々に溜まる場合があります。 この状態を回避するには、タイムアウトをゼロ以外の値に設定し (ゼロの場合はタイムアウトになりません)、タイムアウトに達したときに要求が除去され、キューが冗長な要求でいっぱいにならないようにします。タイムアウトの必要がない場合や タイムアウトになると予想されない場合でも、タイムアウト値を 86400 秒 (24 時間) などの 大きな値に設定すると、ときどきキューから古い集約を 取り除くことができます。
    4. このターミナルに伝搬される入力メッセージ内の要求の分析および明細を提供する 1 つ以上の Compute ノードの In ターミナルに、 AggregateControl ノードの Out ターミナルを接続します。
    重要: AggregateControl ノードの Control ターミナルは、バージョン 6.0 で非推奨になり、デフォルトでは、そのターミナルから AggregateReply ノードへの接続は (直接または間接のいずれも) 無視されます。 この構成は集約フローの効率を最大化し、集約の信頼性を損ないません。 これは、最適な構成です。

    ただし、制御メッセージを AggregateControl ノードから AggregateReply ノードに送信したい場合、制御ターミナルを、ファンイン・フロー上の対応する AggregateReply ノードに (ファンアウトおよびファンイン集約フローの関連付けで説明されているとおり、直接または間接的に) 接続する必要があります。 例えば MQOutput ノードを介する場合などして、AggregateReply ノードに間接的に接続する場合、Compute ノードを組み込んで、該当するヘッダーをメッセージに追加し、安全に送信されることを確かめる必要があります。

    さらに、制御ターミナルとそこからの接続を認識させるには、環境変数 MQSI_AGGR_COMPAT_MODE を使用可能にする必要があります。 ただし、このオプションを選択すると、メッセージ集約のパフォーマンスおよび動作に関係する影響があります。 これらの影響および環境変数の完全な説明については、集約フローでの制御メッセージの使用を参照してください。

    Compute ノード
    Compute ノードは入力メッセージから情報を取り出し、新規の出力メッセージを構成します。

    サブタスク要求を処理するターゲット・アプリケーションが単一の入力メッセージから必要な情報を取り出せる場合、 メッセージを分割するために Compute ノードを組み込む必要はありません。 すべてのターゲット・アプリケーションに入力メッセージ全体を渡すことができます。

    ターゲット・アプリケーションが、入力メッセージ全体ではなく、個々の要求を受信することが予期される場合、 入力メッセージからそれぞれ個々のサブタスク出力メッセージを生成するために Compute ノードを組み込まなければなりません。 それぞれの出力メッセージに入力メッセージの適切なサブセットをコピーして、 以下の方法でそれぞれの Compute ノードを構成してください。

    1. Compute ノードを選択して「プロパティー」ビューをオープンするか、またはノードを右クリックして「プロパティー」をクリックし、「プロパティー」ダイアログを表示します。 ノードのプロパティーが表示されます。
    2. 基本プロパティー「計算モード」の値を選択します。このプロパティーは、ノードによって変更されるメッセージ・ツリーのセクションを指定します。AggregateControl ノードは、 メッセージが到着するとき AggregateRequest ノードが読み取る入力メッセージ内の LocalEnvironment ツリーにエレメントを挿入します。 LocalEnvironment が、Compute ノード内の入力メッセージから出力メッセージに確実にコピーされるようにしてください。 この構成は、LocalEnvironment (「すべて」「LocalEnvironment」「LocalEnvironmentとメッセージ」、または「例外と LocalEnvironment」のうちの 1 つ) を含む値を指定しない限り、自動的に行われます。

      これらの値のいずれかを指定する場合、ブローカーは、ユーザーが LocalEnvironment に書き込む ESQL で Compute ノードをカスタマイズし、 出力メッセージ内で必要なそのツリー内の任意のエレメント上にコピーすると想定します。

      LocalEnvironment を変更したい場合には、以下のステートメントを追加し、入力メッセージから出力メッセージに必要な集約情報をコピーしてください。

      SET OutputLocalEnvironment.ComIbmAggregateControlNode = 
                  InputLocalEnvironment.ComIbmAggregateControlNode;
    3. オプション: このノードのために構成したいその他のプロパティーの値を指定します。
    4. それぞれの Compute ノードの Out ターミナルを、 このノード内の入力メッセージから作成した出力要求メッセージの宛先を表す出力ノードの In ターミナルに接続します。
    Output ノード
    ファンアウト・フロー内に生成するそれぞれの出力メッセージごとに出力ノードを組み込みます。 それぞれの宛先ごとに適切な変更を加え、各ノードを構成してください (その方法については、このセクションで後述します)。

    出力ノードは、MQOutput ノードなどの、要求/応答モデルをサポートする出力ノード、あるいはこれらのノードの混合でなければなりません (これはターゲット・アプリケーションの要件に応じて異なります)。

    1. 出力ノードを選択して「プロパティー」ビューをオープンするか、またはノードを右クリックして「プロパティー」をクリックし、「プロパティー」ダイアログを表示します。 ノードのプロパティーが表示されます。
    2. このノードの出力メッセージの宛先を指定します。 たとえば、MQOutput ノードのメッセージの送信先となる 基本プロパティー「キュー名」内に WebSphere MQ キューの名前を指定します。ターゲット・アプリケーションは、その要求を処理し、 その入力メッセージ内に示されている応答宛先 (たとえば、WebSphere MQ ReplyToQueue) に応答を送信しなければなりません。
    3. 左側のビューで「要求」をクリックし、これらのプロパティーの値を設定して、応答がファンイン・フローの入力キューに送信されることを指定します。
    4. オプション: このノードのために構成したいその他のプロパティーの値を指定します。
    5. 出力ノードの Out ターミナルを、AggregateRequest ノードの In ターミナルに接続します。 メッセージが出力ノードの Out ターミナルを通して伝搬されると、組み込み出力ノードは、 関連する LocalEnvironment 内の WrittenDestination フォルダーを AggregateRequest ノードが必要とする追加情報に更新します。

      組み込みノードが書き込む情報は、キュー名、キュー・マネージャー名、(MQMD からの) メッセージ ID、(MQMD からの) 相関 ID、 およびメッセージ応答 ID (メッセージ ID と同じ値に設定される) です。

    AggregateRequest ノード
    ファンアウト・フロー内に生成するそれぞれの出力メッセージごとに AggregateRequest ノードを組み込みます。
    1. AggregateRequest ノードを選択して「プロパティー」ビューをオープンするか、またはノードを右クリックして「プロパティー」をクリックし、「プロパティー」ダイアログを表示します。 ノードのプロパティーが表示されます。
    2. 基本プロパティー「フォルダー名」を、送信された要求のタイプを識別する値に設定します。この値は、応答メッセージがファンイン・フロー内で受信される際に、 その応答メッセージと一致する AggregateReply ノードにより使用されます。 ファンアウト・フローが生成するそれぞれの要求ごとに指定するフォルダー名は固有でなければなりません。

    AggregateRequest ノードは、 処理するそれぞれのメッセージごとに WebSphere MQ 内にレコードを作成します。 このレコードにより、AggregateReply ノードは、それぞれの応答が関連付けられる要求を識別することができます。 出力ノードが非トランザクションである場合、このデータベースの更新がコミットされる前に、応答メッセージがファンイン・フローに着信することがあります。 この状態を避けるためにタイムアウトを使用する方法の詳細については、集約のタイムアウトの設定を参照してください。

    注意:
    タイムアウトの使用は、前述のこの状況の回避に役立つ可能性がありますが、 ファンアウト・フローをトランザクションに構成することで、対応する AggregateRequest ノードが、そのデータベース更新をコミットする前に、応答メッセージがファンイン・フローに到達することが決してないようにします。
  4. Ctrl+S を押すか、またはタスクバー・メニューで「ファイル」 > 「保存: name (name はメッセージ・フローの名前) をクリックして、メッセージ・フローを保存し、その構成を検証します。
ファンアウト・フローにより開始された集約応答を収集し、ファンイン・フローを作成するには、集約ファンイン・フローの作成を参照してください。
関連概念
メッセージ・フローの概要
ローカル環境のツリー構造
メッセージ・フローの集約
関連タスク
集約フローの構成
集約ファンイン・フローの作成
集約フローでの制御メッセージの使用
ファンアウトおよびファンイン集約フローの関連付け
集約のタイムアウトの設定
複数の AggregateControl ノードの使用
集約フローでの例外の処理
ユーザー定義拡張機能の開発
関連資料
AggregateControl ノード
AggregateReply ノード
AggregateRequest ノード
Compute ノード
MQInput ノード
MQOutput ノード
特記事項 | 商標 | ダウンロード | ライブラリー | サポート | フィードバック

Copyright IBM Corporation 1999, 2009Copyright IBM Corporation 1999, 2009.
最終更新 : 2009-02-20 12:42:50

ac12290_