JBoss Fuseを使い倒す その3:デザインパターン詳細編

2015年9月18日(金)
佐藤 匡剛(さとう ただよし)

集約戦略の実装

メッセージの集約戦略を実装するには、org.apache.camel.processor.aggregate.AggregationStrategyインターフェースを実装するクラスを作成し、その唯一のメソッドであるaggregate(Exchange, Exchange)メソッドを定義します。

以下のクラスは、個々のメッセージの中身(Body)を単純にリストにして集約する戦略の実装例です。

import org.apache.camel.processor.aggregate.AggregationStrategy;

public class ListAggregationStrategy implements AggregationStrategy {

    @Override
    public Exchange aggregate(Exchange oldExchange, Exchange newExchange) {
        if (oldExchange == null) {
            List<String> body = new ArrayList<>();
            body.add(newExchange.getIn().getBody(String.class));
            newExchange.getIn().setBody(body);
            return newExchange;
        }

        List<String> body = oldExchange.getIn().getBody(List.class);
        body.add(newExchange.getIn().getBody(String.class));
        return oldExchange;
    }

}

aggregateメソッド引数のoldExchangeが集約中のメッセージ、newExchangeが新たに届いたメッセージを表します。新しい相関IDを持つ最初のメッセージが届いたときはoldExchangeがnullになっているので、oldExchangeがnullかどうかで処理を条件分岐します。メソッドの戻り値には、集約後のメッセージを返します。

複数のパターンを組み合わせた高度なルーティング

CamelのEIPサポートでは、これまで見てきたRecipient ListやAggregatorのようにDSL上に単一のノードを置くだけで実装が終わってしまうパターンもあれば、Composed Message ProcessorやScatter-Gatherのように本来、複数のパターンを組み合わせて実現すべきより高次のパターンもあります。

最後の締めくくりに、そのような高次のパターンの実装例を1つ取り上げます。

Scatter-Gatherパターン

パターン概要
Scatter-Gather
分散・回収
複数の宛先にメッセージを送信し、その結果をまとめて受信したい場合に、受信者リストまたはパブリッシュサブスクライブ・チャネルを用いてメッセージを一斉配信し、結果を集約器でまとめて受け取る

Scatter-Gatherパターンのアイデアは、これまでに登場した2つのパターンRecipient ListとAggregatorでそれぞれ実現した、メッセージのブロードキャストと集約を組み合わせるというものです。メッセージのブロードキャストと集約を組み合わせると、1つのメッセージを複数の外部サービスに同時に並列処理させ、それぞれの結果を最後にまとめて集計結果を得るといったような、高度なルーティングを実現できます。

Recipient Listパターンのところで議論したように、EIPでメッセージのブロードキャストを実現するにはPublish-Subscribe ChannelとRecipient Listの2通りの方法があり、したがってScatter-Gatherの実現方法も以下の2通りが考えられます。

  1. Publish-Subscribe Channel + Aggregator
  2. Recipient List + Aggregator

この記事では、これまでに登場したRecipient ListとAggregatorを組み合わせる方法を取り上げます。Scatter-Gatherを実装するには、メッセージをブロードキャストするルート(Scatterルート)と、それを集約するルート(Gatherルート)の2つのCamelルートが必要です。それぞれRecipientListノードとAggregateノードを使います。

Scatter-Gatherパターンのブロードキャスト側(Scatter)ルート実装例

図7:Scatter-Gatherパターンのブロードキャスト側(Scatter)ルート実装例

Scatter-Gather パターンの集約側(Gather)ルート実装例

図8:Scatter-Gather パターンの集約側(Gather)ルート実装例

RecipientListとAggregateの各ノードに設定するプロパティは、それぞれRecipient ListパターンとAggregatorパターンのところで説明した内容がそのまま当てはまります。

Scatter-Gather パターン全体の動作は、次の通りです。まず、第一のScatterルートでRecipientListノードから複数のエンドポイントへメッセージが送られます。メッセージを送られた先の個々のエンドポイントは、それがSOAPやRESTのWebサービスであっても、JMSメッセージコンシューマとなるJava EEアプリケーションでも、あるいは別のCamelルートであっても構いません。そのエンドポイントが、受信したメッセージに対して何らかの処理を実行し、その結果が今度はGatherルートのエンドポイント(図8の「seda:gather/in」)へ入力メッセージとして送り返されます。Scatterルートからメッセージをブロードキャストされたどのエンドポイントも、最終的にGatherルートの一点へメッセージを集めるところがポイントです。最後に、GatherルートはAggregateノードに設定されたメッセージ集約のポリシー(Aggregatorパターンの説明を参照)に基づき、メッセージを1つに集約します。

XML DSLによる定義は、以下のようになります。

<route id="EIP-Scatter">
  <from uri="direct:scatter/in" />
  ...
  <recipientList>
    <header>addresses</header>
  </recipientList>
  ...
</route>

<route id="EIP-Gather">
  <from uri="seda:gather/in" />
  ...
  <aggregate strategyRef="aggregationStrategy">
    <correlationExpression>
      <header>aggregateId</header>
    </correlationExpression>
    <completionTimeout>
      <constant>1000</constant>
    </completionTimeout>
    ...
    <to uri="direct:gather/out" />
  </aggregate>
</route>

おわりに

本記事では4つのルーティングパターンだけを取り上げましたが、それだけでもいかに簡単に、かつ保守しやすい形でEIPを実装できるのか、そのJBoss Fuseのパワーを実感いただけたかと思います。実際JBoss Fuseは、60以上あるEIPのほとんどをカバーしています。繰り返しになりますが、それらのEIPを適材適所で使い分けられることが、システム間統合ソリューションの構築を成功させる上で大変重要になります。

その他のEIPの実装方法については、以下のドキュメントを参照してください。
Apache Camel Development Guide - Chapter 3. Introducing Enterprise Integration Patterns

本記事では複数のEIPを一度に取り上げた都合上、これまでの記事のように、1つ1つのパターンについて、ステップ・バイ・ステップでサンプルプロジェクトの作成はしませんでした。その代わりに、紹介した4つのパターンを含んだサンプルプロジェクトをGitHub上に公開いたしました。

https://github.com/tadayosi/thinkit-fuse-eip

以下のように、プロジェクトをローカルにクローンした後、Mavenのtestコマンドを実行することでサンプルコードの動作を確認できます。

$ git clone https://github.com/tadayosi/thinkit-fuse-eip.git
$ cd thinkit-fuse-eip/
$ mvn test

また、クローンしたプロジェクトをJBDSにインポートして閲覧することも可能です。プロジェクトをJBDS上にMavenプロジェクトとしてインポートし([Fileメニュー]→[Import]→[Maven]→[Existing Maven Projects])、その後インポートしたプロジェクトを右クリックして[Enable Fuse Camel Nature]を実行することで、Camelルートの閲覧が可能になります。

著者
佐藤 匡剛(さとう ただよし)
レッドハット株式会社

グローバルサポートサービス JBossシニアソフトウェアメンテナンスエンジニア
オブジェクト指向ソフトウェアアーキテクト、SOA/ESBコンサルタントなどを経験した後、レッドハットに入社。レッドハットでは、JBoss Fuse Service Works、JBoss SOA Platform、JBoss WS、RESTEasyなどのインテグレーションミドルウェア製品に関する技術サポート、バグ修正、上流オープンソースプロジェクト(SwitchYard、JBoss ESB、JBoss WS、Apache Camel、Apache CXF)への貢献を行っている。ドメイン駆動設計(DDD)をはじめとするオブジェクト指向方法論や、システム間統合技術を専門としているが、最近は関数型プログラミングにも興味があり、勉強している。毎朝、娘を幼稚園に送り届けるのが日課。その後、自宅に戻って在宅勤務(Work From Home)をしている。

連載バックナンバー

Think ITメルマガ会員登録受付中

Think ITでは、技術情報が詰まったメールマガジン「Think IT Weekly」の配信サービスを提供しています。メルマガ会員登録を済ませれば、メルマガだけでなく、さまざまな限定特典を入手できるようになります。

Think ITメルマガ会員のサービス内容を見る

他にもこの記事が読まれています