集約ファンアウト・フローは、初期入力メッセージを受け取ると、それを再構築してさまざまなターゲット・アプリケーションへの要求を表示します。
同じメッセージ・フロー内にファンアウトおよびファンイン・フローを組み込むことは可能です。 ただし、2 つの別々のフローを作成してもかまいません。 別個のメッセージ・フローを構成することの利点について詳しくは、ファンアウトおよびファンイン集約フローの関連付けを参照してください。
WebSphere® Message
Broker で提供されているファンアウト・フローの例について検討するには、以下のサンプルを参照してください。
サンプルに関する情報は、WebSphere Message
Broker Toolkit に統合されているインフォメーション・センター、またはオンライン・インフォメーション・センターを使用する場合にのみ表示できます。 サンプルは、WebSphere Message
Broker Toolkit に統合されているインフォメーション・センターを使用する場合にのみ実行できます。
ファンアウト・フローを作成するには、以下のようにします。
- ファンアウト処理を提供するためにメッセージ・フローを作成します。 詳しくは、メッセージ・フローの作成を参照してください。
- 以下のノードをメッセージ・フローに追加してから、説明されているとおりにこれらを構成および接続します。
- 入力ノード
- 入力ノードは、多重要求メッセージが生成元である入力メッセージを受け取ります。 このノードは、組み込みノードのいずれかまたはユーザー定義の入力ノードにすることができます。
- 入力ノードを選択して、「プロパティー」ビューを開きます。 ノードのプロパティーが表示されます。
- このノードの入力メッセージのソースを指定します。 例えば、「キュー名」プロパティーに、WebSphere MQ キューの名前を指定します。
MQInput ノードは、このキューからメッセージを取得します。
- オプション: このノードのために構成したいその他のプロパティーの値を設定します。 例えば、「トランザクション・モード」プロパティーを「はい」に設定し、集約要求メッセージが同期点の下に書き込まれるようにします。 このオプションにより、AggregateReply ノードが、集約インスタンスについて通知する制御メッセージを受け取る前に応答メッセージを受け取るという状況が回避されます。 ファンアウト・フローをトランザクション制御の下に置くことで、
応答メッセージが AggregateReply に到達する前にファンアウト・フローを完了させることができます。
- 入力ノードの Out ターミナルを、AggregateControl ノードの In ターミナルに接続します。 このオプションは、最も単純な構成を表します。適当であれば、入力ノードと AggregateControl ノードとの間に他のノードを組み込むことができます。 例えば、監査のために要求を (Database ノードで) 保管したり、
または固有 ID をメッセージに (Compute ノードで) 追加したりします。
- オプション: ファンアウトおよびファンイン・フローが 1 つのメッセージ・フロー内で結合されている場合は、「拡張」タブの「順序モード」プロパティーを変更します。 「キュー順序」オプションを選択し、「論理順序」プロパティーも選択されていることを確認します。 これらのオプションによって、 キューに到着するメッセージの論理順序が維持されるよう、入力ノードが単一スレッドとなります。 使用可能にする追加のインスタンス・スレッドはすべて、ファンイン入力ノード間でのみ共用され、集約のパフォーマンスが改善されます。 ファンインおよびファンアウト・フローが別々のメッセージ・フローに存在する場合は、追加のスレッドを特にファンイン・フローに対して使用可能にすることができるため、このステップは不要です。
- AggregateControl ノード
- AggregateControl ノードは、入力メッセージに関連するローカル環境を、AggregateRequest ノードが必要とする情報によって更新します。 AggregateControl ノードは、LocalEnvironment.ComIbmAggregateControlNode フォルダーを作成します。
このフォルダーとその中にあるフィールドは、WebSphere Message
Broker が内部で使用するためのものであり、メッセージ・フローを開発する時にそれらのフィールドの存在やそれらの値に依存すべきではありません。
- AggregateControl ノードを選択して、「プロパティー」ビューを開きます。 ノードのプロパティーが表示されます。
- この特定の集約を識別するには、AggregateControl ノードの「集約名」>プロパティーを設定します。 このプロパティーは、後にこの AggregateControl ノードを、
特定の AggregateReply ノードに関連付けるために使用されます。
指定する「集約名」は、ブローカー内でコンテキストに応じて固有でなければなりません。
- オプション: 「タイムアウト」プロパティーを設定して、
応答がアクションを実行するまでブローカーが待機する時間を指定します (集約のタイムアウト値の設定で説明されています)。
AggregateControl でタイムアウトが設定されていない場合、最初に格納された集約要求は、
すべての集約要求応答メッセージが戻されるまで除去されません。 この状態により、内部キューにメッセージが徐々に溜まる場合があります。 この状態を回避するには、タイムアウトをゼロ以外の値に設定し (ゼロの場合は
タイムアウトが決して起きないことを意味します)、タイムアウトに達したときに要求が除去され、
キューが冗長な要求でいっぱいにならないようにします。 タイムアウトの必要がない場合や
タイムアウトになると予想されない場合でも、タイムアウト値を 86400 秒 (24 時間) などの
大きな値に設定すると、ときどきキューから古い集約を
取り除くことができます。
- このターミナルに伝搬される入力メッセージ内の要求の分析および明細を提供する 1 つ以上の Compute ノードの In ターミナルに、
AggregateControl ノードの Out ターミナルを接続します。
重要: AggregateControl ノードの制御ターミナルは、バージョン 6.0 では非推奨でした。そしてデフォルトでは、このターミナルから AggregateReply ノードへの接続は (直接または間接のいずれも) 無視されます。 この構成は集約フローの効率を最大化し、集約の信頼性を損ないません。 これは、最適な構成です。
ただし、制御メッセージを AggregateControl ノードから AggregateReply ノードに送信したい場合、制御ターミナルを、ファンイン・フロー上の対応する AggregateReply ノードに (ファンアウトおよびファンイン集約フローの関連付けで説明されているとおり、直接または間接的に) 接続する必要があります。
例えば MQOutput ノードを介する場合などして、AggregateReply ノードに間接的に接続する場合、Compute ノードを組み込んで、該当するヘッダーをメッセージに追加し、安全に送信されることを確かめる必要があります。
さらに、制御ターミナルとそこからの接続を認識させるには、環境変数 MQSI_AGGR_COMPAT_MODE を使用可能にする必要があります。 ただし、このオプションを選択すると、メッセージ集約のパフォーマンスおよび動作に関係する影響があります。
これらの影響および環境変数の完全な説明については、集約フローでの制御メッセージの使用を参照してください。
- Compute ノード
- Compute ノードは入力メッセージから情報を取り出し、新規の出力メッセージを構成します。
サブタスク要求を処理するターゲット・アプリケーションが単一の入力メッセージから必要な情報を取り出せる場合、
メッセージを分割するために Compute ノードを組み込む必要はありません。 すべてのターゲット・アプリケーションに入力メッセージ全体を渡すことができます。
ターゲット・アプリケーションが、入力メッセージ全体ではなく、個々の要求を受信することが予期される場合、
入力メッセージからそれぞれ個々のサブタスク出力メッセージを生成するために Compute ノードを組み込まなければなりません。 それぞれの出力メッセージに入力メッセージの適切なサブセットをコピーして、
以下の方法でそれぞれの Compute ノードを構成してください。
- Compute ノードを選択して、「プロパティー」ビューを開きます。 ノードのプロパティーが表示されます。
- 基本プロパティー「計算モード」の値を選択します。 このプロパティーは、ノードによって変更されるメッセージ・ツリーのセクションを指定します。 AggregateControl ノードは、メッセージが到着するとき AggregateRequest ノードが読み取る入力メッセージ内のローカル環境ツリーにエレメントを挿入します。 ローカル環境が、Compute ノード内の入力メッセージから出力メッセージに確実にコピーされるようにしてください。 この構成は、ローカル環境 (「すべて」、「LocalEnvironment」、「LocalEnvironment とメッセージ」、または「例外と LocalEnvironment」のうちの 1 つ) を含む値を指定しない限り、自動的に行われます。
これらの値のいずれかを指定する場合、ブローカーは、ユーザーがローカル環境に書き込む ESQL で Compute ノードをカスタマイズし、出力メッセージ内で必要なそのツリー内の任意のエレメントをコピーしていると想定します。
ローカル環境を変更するには、以下のステートメントを追加し、入力メッセージから出力メッセージに必要な集約情報をコピーしてください。
SET OutputLocalEnvironment.ComIbmAggregateControlNode =
InputLocalEnvironment.ComIbmAggregateControlNode;
- オプション: このノードのために構成したいその他のプロパティーの値を設定します。
- それぞれの Compute ノードの Out ターミナルを、
このノード内の入力メッセージから作成した出力要求メッセージの宛先を表す出力ノードの In ターミナルに接続します。
- Output ノード
- ファンアウト・フロー内に生成するそれぞれの出力メッセージごとに出力ノードを組み込みます。 それぞれの宛先ごとに適切な変更を加え、このセクションで後ほど説明されているように各ノードを構成してください。
出力ノードは、MQOutput ノードなどの、要求/応答モデルをサポートする出力ノード、あるいはこれらのノードの混合でなければなりません (これはターゲット・アプリケーションの要件に応じて異なります)。
- 出力ノードを選択して、「プロパティー」ビューを開きます。 ノードのプロパティーが表示されます。
- このノードの出力メッセージの宛先を指定します。
例えば、MQOutput ノードのメッセージの送信先となる「キュー名」プロパティー内に WebSphere MQ キューの名前を指定します。 ターゲット・アプリケーションは、その要求を処理し、
その入力メッセージ内に示されている応答宛先 (例えば、WebSphere MQ ReplyToQueue) に応答を送信しなければなりません。
- 左側のビューで「要求」をクリックし、これらのプロパティーの値を設定して、応答がファンイン・フローの入力キューに送信されることを指定します。
- オプション: このノードのために構成したいその他のプロパティーの値を設定します。
- 出力ノードの Out ターミナルを、AggregateRequest ノードの In ターミナルに接続します。
メッセージが出力ノードの Out ターミナルを通して伝搬されると、組み込み出力ノードは、
関連するローカル環境内の WrittenDestination フォルダーを AggregateRequest ノードが必要とする追加情報に更新します。
組み込みノードが書き込む情報は、キュー名、キュー・マネージャー名、(MQMD からの) メッセージ ID、(MQMD からの) 相関 ID、
およびメッセージ応答 ID (メッセージ ID と同じ値に設定される) です。
- AggregateRequest ノード
- ファンアウト・フロー内に生成するそれぞれの出力メッセージごとに AggregateRequest ノードを組み込みます。
- AggregateRequest ノードを選択して、「プロパティー」ビューを開きます。 ノードのプロパティーが表示されます。
- 基本プロパティー「フォルダー名」を、送信された要求のタイプを識別する値に設定します。 この値は、応答メッセージがファンイン・フロー内で受信される際に、
その応答メッセージと一致する AggregateReply ノードにより使用されます。
ファンアウト・フローが生成するそれぞれの要求ごとに指定するフォルダー名は固有でなければなりません。
AggregateRequest ノードは、
処理するそれぞれのメッセージごとに WebSphere MQ 内にレコードを作成します。 このレコードにより、AggregateReply ノードは、それぞれの応答が関連付けられる要求を識別することができます。 出力ノードが非トランザクションである場合、このデータベースの更新がコミットされる前に、応答メッセージがファンイン・フローに着信することがあります。 この状態を避けるためにタイムアウトを使用する方法の詳細については、集約のタイムアウト値の設定を参照してください。
注意:
タイムアウトの使用は、前述のこの状況の回避に役立つ可能性がありますが、ファンアウト・フローをトランザクションに構成してください。 そのようにして、対応する AggregateRequest ノードが、データベース更新をコミットする前に、応答メッセージがファンイン・フローに到達することがないようにします。
- メッセージ・フローを保存し、その構成を検証するには、Ctrl-S を押すか、またはをクリックします。
集約ファンイン・フローの作成で説明されているように、ファンアウト・フローにより開始された集約応答を収集するには、ファンイン・フローを作成します。