Muleフローのルーティング処理を使いこなそう

今回のテーマはルーティング。分岐とちょっと似ていますが使い方はかなり異なるので特徴を押さえておきましょう。

今回使うコンポーネント
ルーティングを行うコンポーネント
1. Round Robin
2. Scatter-Gather
まとめ
1. いいね:

今回使うコンポーネント

コンポーネント	アイコン	用途
Round Robin		複数ルートを設定して、実行回数によりルーティングを行う
Scatter-Gather		処理を分割して最後に同期処理を行う

ルーティングを行うコンポーネント

Round Robin

Round Robinはフロー内にルートを複数設定し、順番によって通るルートを変えていくような仕組みとなります。条件ではなく、回数によってルートが変わっていくため、ちょっと使い所が難しいかもしれません。

実際にRound Robinのフローを組んでみたいと思います。今回は複数回通すようにFor文の中にRound Robinを入れるような実装にします。

①For each文には配列を渡します。今回は4打席分用意します。結果は残念ならら4三振です。

②〜④Round Robinの各ルートにはそれぞれ異なる打撃成績をセットします。②にはpayload、③には「四球」、④には「ホームラン」をセットします。
⑤打撃結果を配列に追加します。細かい処理の説明については割愛しますが、以下のようにDataweaveを記載し、batting変数に打撃成績が追加されるようにします。

%dw 2.0
var result = "第" ++ vars.counter ++ "打席は" ++ payload ++ "でした"
output application/json
---
vars.batting default [] ++ [result]

⑥最後に変数battingの内容をPayloadに格納します。

%dw 2.0
output application/json
---
vars.batting

上記フローで実行した結果はこのようになります。

このように配列4要素に対し、1回目と4回目は配列の内容がそのまま設定され、2回目と3回目がそれぞれ値が書きかわっていることが分かります。Round Robinを通る回数によってそれぞれ通るルートが異なることが分かっていただけたでしょうか。そして、ルートの数を超えた分はまた最初のルートのものが実行されるような仕組みになります。

Scatter-Gather

Scatter-Gatherはルートを複数分岐させ、各ルートの完了後にまた1つのフローへと合流する仕組みとなっています。感覚的にはParallel For Eachに似ており、繰り返し処理の部分を好きなように設定できる形となります。

実際にScatter-Gatherのフローを組んで動きを確かめてみましょう。こちらのフローでは、大谷選手の投手と野手の成績をそれぞれ表示させてみたいと思います。以下のようにフローを配置し、Scatter-Gatherの各ルートに投手成績と野手成績を入れていきます。最後にScatter-Gather直後のPayload値をJSON形式で出力します。

Scatter-Gather
・複数ルートを並列処理する
・全ルートの処理完了後に同期が取られ、各処理のPayloadが集約される
・ルート内はパラレル実行などので、ルート内で更新した変数が別ルートに反映されない
・負荷分散に使う

フローを実行した結果がこちらになります。ちょっと長いのでキャプチャを取るより、テキストベースで貼り付けます。

{
  "0": {
    "inboundAttachmentNames": [],
    "exceptionPayload": null,
    "inboundPropertyNames": [],
    "outboundAttachmentNames": [],
    "outboundPropertyNames": [],
    "payload": "大谷選手がホームランを打ちました",
    "attributes": {
      "headers": {
        "host": "localhost:8081"
      },
      "clientCertificate": null,
      "method": "GET",
      "scheme": "http",
      "queryParams": {},
      "requestUri": "/scatter_gather",
      "queryString": "",
      "version": "HTTP/1.1",
      "maskedRequestPath": null,
      "listenerPath": "/scatter_gather",
      "localAddress": "/127.0.0.1:8081",
      "relativePath": "/scatter_gather",
      "uriParams": {},
      "rawRequestUri": "/scatter_gather",
      "rawRequestPath": "/scatter_gather",
      "remoteAddress": "/127.0.0.1:49598",
      "requestPath": "/scatter_gather"
    }
  },
  "1": {
    "inboundAttachmentNames": [],
    "exceptionPayload": null,
    "inboundPropertyNames": [],
    "outboundAttachmentNames": [],
    "outboundPropertyNames": [],
    "payload": "大谷選手が7回無失点でした",
    "attributes": {
      "headers": {
        "host": "localhost:8081"
      },
      "clientCertificate": null,
      "method": "GET",
      "scheme": "http",
      "queryParams": {},
      "requestUri": "/scatter_gather",
      "queryString": "",
      "version": "HTTP/1.1",
      "maskedRequestPath": null,
      "listenerPath": "/scatter_gather",
      "localAddress": "/127.0.0.1:8081",
      "relativePath": "/scatter_gather",
      "uriParams": {},
      "rawRequestUri": "/scatter_gather",
      "rawRequestPath": "/scatter_gather",
      "remoteAddress": "/127.0.0.1:49598",
      "requestPath": "/scatter_gather"
    }
  }
}

各ルートの処理結果がインデックスされてそれぞれ出力されていることが分かります。今回は2ルートでしたが、ルートを増やせば、0,1,2…とインデックスが作られそれぞれに結果が入る仕組みとなっています。