5 チャネル

チャネルは、他のタイプのコンポーネント(ステージ)間をイベントが往来する論理的なパイプを表します。たとえば、アダプタとOracle CQLプロセッサ間やOracle CQLプロセッサとイベントBean間などがあります。

この章の内容は次のとおりです。

5.1 チャネルを使用するタイミング

チャネルには、バッファリング機能、キューイング機能および同時実行性機能が用意されており、これにより、設計ライフサイクルの後半でアプリケーションのパフォーマンスをチューニングできるようになります。

デフォルトでは、チャネルのmax-threads属性は0に設定されており、これは、チャネルがパススルー・モードであり、パフォーマンス・ペナルティが起きないことを意味します。

EPNを構成するとき、次のルールを考慮してください。

Oracle CQLプロセッサをダウンストリーム・ステージに接続する場合、チャネルは必須です。
ストリームまたはリレーションをOracle CQLプロセッサに接続する場合、チャネルは必須です。

前述の2点に基づいて、アダプタとプロセッサ間でチャネルを設定することは必須であることに注意してください。Oracle JDeveloperを使用してアダプタをプロセッサに接続する場合、チャネルを作成するためのウィザードが表示されます。『Oracle Event Processingスタート・ガイド』のチャネルの作成に関する項を参照してください。
外部リレーション、キャッシュまたは表ソースのいずれかのコンポーネントをOracle CQLプロセッサに接続する場合、チャネルはオプションです。

チャネルは、キャッシュや表などのプル・ソースとプロセッサの間では、プル・ソースが外部リレーションを示すため必要ありません。外部リレーションでは、有効な操作はストリームとNOWウィンドウ演算子間の結合のみとなるため、プル・ソースとみなされます。結合は実際には、Oracle CQLプロセッサの外部で発生します。これはプルであるため、Oracle CQLプロセッサはその状態を認識される必要はなく(つまり、DDLは必要ありません)、チャネルで仲介する必要もありません。

通常、コンポーネント間のチャネルは次の場合に使用します。

バッファリングは、発行するコンポーネントと受信側の間で必要です。
受信側のコンポーネントにはキューイングまたは同時実行性が必要です。
カスタム・アダプタが使用される場合は、スレッド制御が必要です。

5.2 チャネル構成

イベント処理ネットワーク(EPN)にチャネルを追加する場合は、デフォルト構成を使用できます。デフォルトのチャネルは、名前およびIDを持ち、システム・タイムスタンプが付加されたストリーム・チャネルです。デフォルトのハートビートのタイムアウトは100ミリ秒または100,000,000ナノ秒です。

ほとんどのアプリケーションにはデフォルトの構成が適しています。構成を変更するには、アプリケーションのアセンブリ・ファイルを編集するか、コンポーネントの構成ファイルを編集します。

システムによってチャネルにタイムスタンプが付加されている場合、新規イベントが到着したとき、および構成可能なハートビートのタイムアウトが期限切れになったときに、CPUクロックから新しい時間が割り当てられます。

アプリケーションによってチャネルにタイムスタンプが付加されている場合、イベントのタイムスタンプはwlevs:expression要素によって決定されます。標準的な式の例として、イベントでのプロパティへの参照があります。式が指定されていない場合、タイムスタンプは前のイベントから伝播されます。たとえば、1つのOracle CQLプロセッサのシステムによりタイムスタンプが付加されたチャネルが、別のダウンストリームOracle CQLプロセッサのアプリケーションによりタイムスタンプが付加されたチャネルにイベントを送信する場合などです。また、アプリケーションはStreamSender.sendHeartbeatメソッドを使用して、イベント型heart-beatダウンストリームをEPNのStreamSinkリスナーに送信することが可能です。

注:

チャネルがアプリケーションによってタイムスタンプが付加されていると同時にマップ・ベースである(ハッシュ・マップ・イベント・タイプを使用する)場合、タイムスタンプが追加されます。キーのない削除または更新操作はこの構成のチャネルに対しては機能しません。これは、アプリケーションによりタイムスタンプが付加されたイベントによって常にtimestampプロパティの変更が回避されるためです。

この章では、いくつかのアセンブリおよび構成ファイルのチャネル設定について説明します。

5.2.1 アセンブリ・ファイル

アセンブリ・ファイルは、helloworldInputChannelのチャネル設定を示します。この設定は、helloworldProcessorがイベントのチャネルをリスニングし、このイベントがhelloworldAdapterからチャネルにフローすることを示します。

<wlevs:channel id="helloworldInputChannel" event-type="HelloWorldEvent" >
   <wlevs:listener ref="helloworldProcessor"/>
   <wlevs:source ref="helloworldAdapter"/>
</wlevs:channel>

チャネルをリレーションとして構成するには、次のように、is-relation設定をアセンブリ・ファイルに追加します。

<wlevs:channel id="helloworldInputChannel" event-type="HelloWorldEvent"
is-relation="true" primary-key="myprimarykey" />

このチャネルをリレーションとして指定する場合は、primary-key属性も構成する必要があります。主キーは、各イベントを一意に識別する空白またはカンマ区切りのイベント・プロパティ名のリストです。プライマリ・キーを定義する方法の詳細は、『Oracle Event Processing スキーマ・リファレンス』のwlevs:metadataに関する項を参照してください。

アプリケーションによってタイムスタンプを付加するチャネルを構成するには、次のように、application-timestampedおよびexpression要素をアセンブリ・ファイルに追加します。is-total-order要素をtrueに設定すると、パブリッシュされたアプリケーション時間が常に前回使用された値よりも大きくなります。

<wlevs:application-timestamped is-total-order="true">
   <wlevs:expression>mytime+10</wlevs:expression>
</wlevs:application-timestamped>

5.2.2 構成ファイル

構成ファイルは、チャネル構成の設定を示します。この設定では、プロセス・イベントの非同期バッファリング(max-size)、最大4つのスレッドの使用(max-threads)、および10000ナノ秒のハートビート・タイムアウトの使用(heartbeat)を行うためにチャネルをカスタマイズします。

  <channel>
    <name>helloworldInputChannel</name>
    <max-size>10000</max-size>
    <max-threads>4</max-threads>
    <heartbeat>10000</name>
</channel>

5.3 ダウンストリーム・チャネルに出力する問合せの制御

複数の問合せを使用してOracle CQLプロセッサを構成する場合、デフォルトではすべての問合せの結果がダウンストリーム・チャネルに送信されます。どの問合せの結果をダウンストリーム・チャネルに送信するかは、selector要素を使用して制御できます。

図5-1は、アップストリームOracle CQLプロセッサfilteredFanoutProcessorに接続されたチャネルfilteredStreamを備えたEPNを示します。

図5-1 Oracle CQLプロセッサおよびダウンストリーム・チャネルを備えたEPN

「図5-1 Oracle CQLプロセッサおよびダウンストリーム・チャネルを備えたEPN」の説明

次の例は、Oracle CQLプロセッサに構成された問合せを示します。

<processor>
    <name>filterFanoutProcessor</name>
    <rules>
        <query id="Yr3Sector"><![CDATA[ 
            select cusip, bid, srcId, bidQty, ask, askQty, seq 
            from priceStream where sector="3_YEAR"
        ></query>
        <query id="Yr2Sector"><![CDATA[ 
            select cusip, bid, srcId, bidQty, ask, askQty, seq 
            from priceStream where sector="2_YEAR"
        ></query>
        <query id="Yr1Sector"><![CDATA[ 
            select cusip, bid, srcId, bidQty, ask, askQty, seq 
            from priceStream where sector="1_YEAR"
        ></query>
    </rules>
</processor>

Oracle CQLプロセッサに複数の問合せを指定する場合、デフォルトでは、すべての問合せ結果がOracle CQLプロセッサのアウトバウンド・チャネルに出力されます(図5-1のfilteredStream)。オプションで、コンポーネント構成ソースでchannel要素のselector子要素を使用し、チャネルに結果を出力する1つ以上のOracle CQL問合せ名のスペース区切りのリストを指定できます。次の例では、問合せYr3SectorおよびYr2Sectorの問合せ結果はfilteredStreamに出力されますが、問合せYr1Sectorの問合せ結果は出力されません。

<channel>
    <name>filteredStream</name>
    <selector>Yr3Sector Yr2Sector</selector>
</channel>

アップストリームOracle CQLプロセッサに問合せを作成する前に、selectorでchannel要素を構成できます。この場合、selector内の名前と一致する問合せ名を指定する必要があります。

注:

selector子要素は、アップストリーム・ステージがOracle CQLプロセッサである場合にのみ適用できます。詳細は、Oracle CQLプロセッサを参照してください。

5.4 バッチ処理チャネル

デフォルトでは、チャネルはイベントが到達した時点でイベントを処理します。wlevs:channel属性batchingをtrueに設定することで、同じタイムスタンプを持ち、同じ問合せからの出力のイベントをまとめてバッチ処理するように構成できます。

イベントのバッチ処理により、アプリケーションのパフォーマンスを向上させることができます。

<wlevs:channel id="priceStream" event-type="PriceEvent" batching="true">
    <wlevs:listener ref="filterFanoutProcessor" />
    <wlevs:source ref="PriceAdapter" />
</wlevs:channel>

5.5 フォルト処理

チャネルのダウンストリーム・ステージで発生し、チャネルにスローされる例外を処理するコードを書き込むことができます。

デフォルトでは、チャネルのフォルト処理動作は次のとおりです。

チャネルのmax-threads設定が0である場合(パススルー・チャネル)、例外がEPNの次のアップストリーム・ステージに再スローされます。
チャネルのmax-threads設定が0より大きい場合、例外がログに記録および削除されます。また、フォルトに関連付けられているイベントもログに記録および削除されます。

フォルト処理クラスを書き込み、max-threads値が0より大きいチャネルとハンドラを関連付けることができます。フォルト・ハンドラをチャネルに関連付けると、チャネルにスローされる例外は、フォルトを処理したり再スローするコードが含まれるハンドラで受信されます。フォルト処理コードが例外を再スローする場合、例外はログに記録されますが例外に関連するイベントは失われます。これらの例外に含まれるイベントを追跡する場合、イベント・データをEPNに接続されるデータ・ソースに書き込むことなどによって、コードを使用して永続化する必要があります。

注:

マルチスレッドのチャネルによってスローされた例外を処理するには、プロセッサなどのチャネルのアップストリームにあるコンポーネントにフォルト・ハンドラが登録されている必要があります。アップストリームのコンポーネントにフォルト・ハンドラを登録しない場合、例外はアップストリームに渡されますが、フォルト・ハンドラは呼び出されません。

フォルト・ハンドラの書込みの詳細は、フォルト処理を参照してください。

5.6 EventPartitionerチャネル

デフォルトでは、チャネルは各リスナーに各イベントをブロードキャストします。

EventPartitionerを使用するようにチャネルを構成した場合、受信イベントが到着するたびにチャネルはリスナーを選択し、各リスナーに各イベントをブロードキャストするかわりに、そのリスナーにイベントをディスパッチします。チャネルでEventPartitionerを使用してスケーラビリティを向上できます。