| Oracle® Fusion Middleware Oracle Event Processingアプリケーションの開発 12c リリース1 (12.1.3) E54312-04 |
|
 前へ |
 次へ |
アプリケーションがイベント・データにアクセスできるようにキャッシング・システムを構成できます。Oracle Coherence分散キャッシング、Oracle Event Processingローカル・キャッシングおよびサードパーティ提供のキャッシング・ソリューションを組み合せてシステムのキャッシュを構成できます。Oracle CQLおよびJavaクラスを使用してキャッシュのイベントにアクセスできます。
この章の内容は次のとおりです。
キャッシュはイベント・データの一時記憶域です。イベント・データの可用性およびアプリケーションのパフォーマンス向上のため、キャッシュを作成できます。アプリケーションでイベントを作成したキャッシュにパブリッシュ、または作成したキャッシュからイベントを使用できるようになります。
また、他のアプリケーションがキャッシュに書き込んだ処理済のイベント・データにもアクセスできます。
イベントを生成するOracle Event Processingアプリケーションで任意のステージを構成し、イベントをキャッシュにパブリッシュできます。キャッシュはEPNのステージである必要はありません。キャッシングAPIを使用してプログラミングすることにより、別のコンポーネントまたはSpring Beanからキャッシュのイベントにアクセスできます。
キャッシング・システムとは、キャッシュ実装の構成されたインスタンスです。キャッシング・システムは、構成されたキャッシュの名前付きセットを定義するとともに、任意のキャッシュが複数のマシンに渡って配信される場合、リモート通信用の構成を定義します。
Oracle Event Processingキャッシングにより、アプリケーションで次のタスクを実行できます。これらのタスクはすべてインクリメンタルに行われ、アプリケーションの停止や待機時間スパイクは発生しません。
アプリケーションのデプロイ前にキャッシュにイベント・データをあらかじめロードします。
キャッシュのイベント・データを定期的にリフレッシュ、無効化およびフラッシュします。
動的にキャッシュ構成を更新します。
Oracle Event Processingは、次のキャッシュ実装をサポートします。
Oracle Event Processingローカル・キャッシュ: メモリー内のローカル・シングルJVMキャッシュ。この実装はローカルの使用に最適です(クラスタで使用できません)。設定が比較的簡単なため、早期の開発にも役立つ場合があります。
Oracle Coherence: クラスタリングされたアプリケーションおよびアプリケーション・サーバーのJCacheに準拠したインメモリー分散型データ・グリッド・ソリューション。これは、クラスタ全体の同時実行性制御を使用してデータへの更新を連携動作させ、使用可能な最高性能のクラスタ化プロトコルを使用してデータ変更をクラスタ全体にレプリケートする他、データ変更通知をそれをリクエストするすべてのサーバーに配信します。ユーザーは、標準JavaコレクションAPIを使用してデータへのアクセスや変更を行うことでOracle Coherenceの機能を活用する他、JavaBeanの標準イベント・モデルを使用してデータ変更通知を受け取ります。
注:
Oracle Event ProcessingをOracle Coherenceと組み合せて使用するには、Coherence Enterprise Edition、Coherence Grid Edition、Oracle WebLogic Application Gridのライセンスなどの有効なOracle Coherenceライセンスを取得する必要があります。
Oracle Coherenceの詳細は、http://docs.oracle.com/middleware/1213/coherence/index.htmlを参照してください。
サードパーティのキャッシュ: Oracle Event Processingをサードパーティの他のキャッシュ実装で機能することを可能にするプラグインを作成できます。
高いスループットが特に重要になる可能性がある場合、キャッシュ技術はストリーミング・データのユース・ケースに非常に役立ちます。通常、キャッシュからのデータの取得は、リレーショナル・データベースからの同じデータの取得よりはるかに高速になります。
Oracle Event Processingアプリケーションでキャッシュする共通のユース・ケースのシナリオは次のとおりです。
キャッシュへのイベントのパブリッシュ
金融アプリケーションは、金融市場が開いている間はイベントをキャッシュにパブリッシュし、市場終了後にキャッシュのデータを処理します。イベントをキャッシュにパブリッシュすると、イベントがアプリケーションまたはサーバー上で稼働する他のOracle Event Processingアプリケーションで使用可能になります。イベントをキャッシュにパブリッシュすると、キャッシュの実装によってセカンダリ・ストレージへの非同期的書込みも可能になります。
キャッシュ内のデータの使用
Oracle Event Processingアプリケーションでは、ストリーミング以外のデータへのアクセスが必要になる場合があります。このデータをキャッシュすることで、アプリケーションのパフォーマンスが向上します。プログラムによりキャッシュに直接アクセスできるOracle Event Processingアプリケーションの標準のコンポーネントは、入力および出力アダプタ、およびビジネスPOJOです。
また、アプリケーションは、ユーザー定義関数またはOracle CQL文から直接のいずれの方法でも、Oracle CQLからキャッシュにアクセスできます。ユーザー定義関数の場合、プログラマはSpringを使用してキャッシュ・リソースを関数の実装に組み込みます。詳細は、アプリケーションおよびリソース構成を参照してください。
アプリケーションは、プロセッサで実行するOracle CQL文から直接キャッシュに問合せを実行することもできます。この場合、データがキャッシュから抽出される場合を除いて、キャッシュへの問合せがチャネルへの問合せに類似するように、キャッシュはプロセッサへの別の種類のデータ・ソースとして機能します。
キャッシュに問い合せるためにOracle CQLを使用する例は、注文や注文をキャッシュに実行するために使用される取引をパブリッシュする金融アプリケーションにみられます。市場が終了する1日の終わりに、アプリケーションはキャッシュへの問合せを実行し、特定の注文に関連するすべての取引を検索します。
キャッシュのデータの更新と削除
Oracle Event Processingアプリケーションでは、必要に応じてキャッシュのデータを更新および削除できます。たとえば、金融アプリケーションでは注文を実現する個々の取引が実行されるたびにキャッシュ内の注文を更新し、注文が取り消された場合はこれを削除する必要があります。キャッシュ内のデータの使用が許可されたアプリケーションのコンポーネントには、データの更新も許可されます。
マルチサーバー・ドメインにおけるキャッシュの使用
キャッシュを使用するOracle Event Processingアプリケーションを構築する場合、そのアプリケーションをマルチサーバー・ドメインでデプロイし、分散型キャッシュをサポートするキャッシング・システムを使用する必要があります。この場合、Oracle Coherenceまたは分散型キャッシュをサポートするサードパーティのキャッシング・システムのいずれかを使用する必要があります。
詳細は、次を参照してください。
Oracle Coherenceキャッシング・システムおよびキャッシュを使用するようにアプリケーションを構成できます。アプリケーションをマルチサーバー・ドメインにデプロイしようとする場合は、このキャッシング・システムを使用します。
Oracle Coherenceを使用して構成すると、第1キャッシュ・システムのみがサーバー内に構成されます。Oracle Event Processingサーバーは、構成済の他のキャッシング・システムを無視します。
注:
Oracle CoherenceとともにOracle Event Processingを正式に使用する前に、Coherence Enterprise Edition、Coherence Grid Edition、またはOracle WebLogic Application Grid用ライセンスなどの有効なCoherenceライセンスを取得する必要があります。
Oracle Coherenceの詳細は、http://docs.oracle.com/middleware/1213/coherence/index.htmlを参照してください。
次のアセンブリおよび構成ファイル設定で、Oracle Coherenceキャッシング・システムおよびOracle CQLプロセッサのキャッシュが構成されます。このキャッシュは、イベント・タイプを使用してリレーショナル・データベースの表の行を特定するキー・プロパティを指定します。このキャッシング・システムは公開されており、他のアプリケーションからキャッシュのデータにアクセスできます。
アセンブリ・ファイル設定により、キャッシング・システムおよびcache1を構成します。value-type設定では、データベースの値をロードするイベント・タイプを指定します。このキャッシュは通知されます。
<wlevs:cache id="cache1" value-type="TradeReport" advertise="true">
<wlevs:caching-system ref="coherence-caching-system"/>
</wlevs:cache>
<wlevs:caching-system id="coherence-caching-system" provider="coherence"/>
注:
EPN図のcoherenceキャッシュのid設定を変更すると、アセンブリ・ファイルとcoherence-cache-ファイルのidが変更されます。ただし、アセンブリ・ファイルのソース・エディタでid設定を変更すると、アセンブリ・ファイルのidのみが変更されます。この場合は、coherence-cache-のcache-name設定を、アセンブリ・ファイルのid設定と一致するように手動で変更する必要があります。また、そのキャッシュへのすべての参照も変更する必要があります。
キャッシュが通知される場合、別のバンドル内にあるアプリケーションのEPNにあるコンポーネントがそのキャッシュを参照できます。次の例では、1つのバンドル内のプロセッサがcache-source要素を使用して、cacheproviderのcache-idを持つ別のバンドルのキャッシュ・ソースを参照する方法を示しています。
<wlevs:processor id="myProcessor2"> <wlevs:cache-source ref="cacheprovider:cache-id"> </wlevs:processor>
注:
同一のOracle Event Processingサーバーにデプロイされる1つ以上のアプリケーションのEPNアセンブリ・ファイルのOracle Coherenceキャッシュが存在する場合、ローダーまたはストアを使用して同一のキャッシュの複数のインスタンスを構成しないでください。
各EPNアセンブリ・ファイル内でローダーまたはストアを使用して同一のOracle Coherenceキャッシュをそれぞれ構成する、複数のアプリケーションを使用することによって、これを不注意で行う可能性があります。ローダーまたはストアを使用して同一のキャッシュの複数のインスタンスを構成すると、Oracle Event Processingで例外をスローします。
coherence-cache-config.xmlファイルは基本的なOracle Coherence構成ファイルで、任意のOracle Coherenceアプリケーションでtrueとなるように、Oracle Coherence DTDに準拠する必要があります。
coherence-cache-config.xml:の詳細は、Oracle Coherenceドキュメント(http://docs.oracle.com/middleware/1213/coherence/index.html)を参照してください。
Springを使用してキャッシュのローダーまたはストアを構成する場合は、Oracle Event Processing Oracle Coherenceファクトリを宣言する必要があります。ファクトリを指定するには、cachestore-scheme要素を使用し、Oracle CoherenceがOracle Event Processingにコールして、キャッシュのために構成されるローダーまたはストアへの参照を取得できるようにするためのファクトリ・クラスを含めます。ローダーまたはストアの構成における唯一の差異は、method-name要素が、ローダーが使用されるときはgetLoaderの値を持ち、ストアが使用されているときはgetStoreの値を持つことです。キャッシュ名は、入力パラメータとしてファクトリに渡します。
<cache-config>
<caching-scheme-mapping>
<cache-mapping>
<cache-name>myCoherenceCache</cache-name>
<scheme-name>new-replicated</scheme-name>
</cache-mapping>
<cache-mapping>
<cache-name>myLoaderCache</cache-name>
<scheme-name>test-loader-scheme</scheme-name>
</cache-mapping>
<cache-mapping>
<cache-name>myStoreCache</cache-name>
<scheme-name>test-store-scheme</scheme-name>
</cache-mapping>
<cache-mapping>
<cache-name>
cache1
</cache-name>
<scheme-name>
new-replicated
</scheme-name>
</cache-mapping>
</caching-scheme-mapping>
<caching-schemes>
<replicated-scheme>
<scheme-name>new-replicated</scheme-name>
<service-name>ReplicatedCache</service-name>
<backing-map-scheme>
<class-scheme>
<scheme-ref>my-local-scheme</scheme-ref>
</class-scheme>
</backing-map-scheme>
</replicated-scheme>
<class-scheme>
<scheme-name>my-local-scheme</scheme-name>
<class-name>com.tangosol.net.cache.LocalCache</class-name>
<eviction-policy>LRU</eviction-policy>
<high-units>100</high-units>
<low-units>50</low-units>
</class-scheme>
<local-scheme>
<scheme-name>test-loader-scheme</scheme-name>
<eviction-policy>LRU</eviction-policy>
<high-units>100</high-units>
<low-units>50</low-units>
<!-- A cachestore-scheme element that gets a loader starts here -->
<cachestore-scheme>
<class-scheme>
<class-factory-name>com.bea.wlevs.cache.coherence.configuration.SpringFactory
</class-factory-name>
<method-name>getLoader</method-name>
<init-params>
<init-param>
<param-type>java.lang.String</param-type>
<param-value>myCoherenceCache</param-value>
</init-param>
<init-param>
<param-type>
java.lang.String
</param-type>
<param-value>
cache1
</param-value>
</init-param>
</init-params>
</class-scheme>
</cachestore-scheme>
<!-- The cachestore-scheme element ends here -->
</local-scheme>
<local-scheme>
<scheme-name>test-store-scheme</scheme-name>
<eviction-policy>LRU</eviction-policy>
<high-units>100</high-units>
<low-units>50</low-units>
<!-- A cachestore-scheme element that gets a store starts here -->
<cachestore-scheme>
<class-scheme>
<class-factory-name>com.bea.wlevs.cache.coherence.configuration.SpringFactory
</class-factory-name>
<method-name>getStore</method-name>
<init-params>
<init-param>
<param-type>java.lang.String</param-type>
<param-value>myCoherenceCache</param-value>
</init-param>
<init-param>
<param-type>
java.lang.String
</param-type>
<param-value>
cache1
</param-value>
</init-param>
</init-params>
</class-scheme>
</cachestore-scheme>
<!-- The cachestore-scheme element ends here -->
</local-scheme>
</caching-schemes>
</cache-config>
tangosol-coherence-override.xmlファイル(オプション)
tangosol-coherence-override.xmlファイルはサーバー単位でグローバルなファイルです。これには、Oracle Coherenceドキュメントで操作構成と呼ばれているものが含まれています。このファイルには、Oracle Coherenceキャッシュのサーバー単位のグローバルな構成設定が含まれています。このファイルをXMLエディタで作成し、構成するOracle Event Processingサーバーのconfigディレクトリに置きます。
注:
Oracle Coherenceをクラスタリングに使用する場合は、tangosol-coherence-override.xmlファイルを含めないでください。
次のXMLをOracle Coherence構成ファイルに追加し、tangosol-coherence-override.xmlファイルを参照します。cluster-name要素を含めることにより、Oracle Event Processingの起動時にOracle Coherenceと既存のOracle Coherenceクラスタの結合が試行されることを回避します。これが発生すると、問題が発生してOracle Event Processingが起動できなくなる場合があります。
...
<coherence xml-override="/tangosol-coherence-override.xml">
<cluster-config>
<member-identity>
<cluster-name>com.bea.wlevs.example.provider</cluster-name>
</member-identity>
...
</coherence>
Oracle Event Processingクラスタの詳細は『Oracle Event Processingの管理』のOracle Event Processingのネイティブ・クラスタに関する項を参照してください。
com.oracle.cep.cacheloader パッケージは、CSVイベントをCoherenceキャッシュにロードするためのCsvCacheLoaderクラスを提供します。sourceUrlプロパティを置き換え、インバウンド・アダプタを持つキャッシュ・ローダーを使用します。
最初のアセンブリ・ファイルCSVアダプタ構成は、sourceUrlプロパティを使用してファイルをロードするCSVインバウンド・アダプタを示します。2番目のアセンブリ・ファイルCSVアダプタ・エントリは、キャッシュ・ローダーBeanをロードするCSVインバウンド・アダプタを示します。
CSVファイルのイベントのロード
<wlevs:adapter id="StockTradeCSVInboundAdapter" provider="csv-inbound">
<wlevs:listener ref="AdapterOutputChannel"/>
<wlevs:instance-property name="eventType" value="TradeEvent"/>
<wlevs:instance-property name="sourceUrl"
value="file:/scratch/mpawlan/oep9-19/oep/utils/load-generator/StockData.csv"/>
</wlevs:adapter>
キャッシュ・ローダーを使用するイベントのロード
<wlevs:cache id="csvcache" key-properties="sequenceNo"
value-type="TradeEvent" advertise="true">
<wlevs:caching-system ref="cachesys" />
</wlevs:cache>
<bean id="csvloader" class="com.oracle.cep.cacheloader.CsvCacheLoader">
<property name="cacheName" value="csvcache"/>
<property name="sourceUrl"
value="file:///scratch/juhe/view_storage/trade.csv"/>
</bean>
Oracle Event Processingローカル・キャッシュ・システムおよびキャッシュを使用するようにアプリケーションを構成できます。Oracle Event Processingローカル・キャッシュ・システムは、アプリケーションをマルチサーバー・ドメインにデプロイしない場合に最適です。
アプリケーションをマルチサーバー・ドメインにデプロイする計画がある場合は、Oracle Coherenceキャッシュを使用します。
この章では、いくつかの構成設定について説明します。詳細は、『Oracle Event Processing スキーマ・リファレンス』のコンポーネント構成スキーマに関する項およびCoherenceキャッシュ・システムに関する項を参照してください。
次のアセンブリ・ファイル設定により、ローカル・キャッシング・システムおよびキャッシュを構成します。value-type設定では、データベースの値をロードするイベント・タイプを指定します。
<wlevs:cache id="localcache" value-type="HelloWorldEvent">
<wlevs:caching-system ref="caching-system"/>
</wlevs:cache>
<wlevs:caching-system id="caching-system" provider="wlevs" advertise="false"/>
次の構成ファイル設定では、ローカル・キャッシング・システムの最大サイズおよび削除ポリシーを指定します。最大サイズは、削除ポリシー発生後のメモリー内のキャッシュ要素数を指定します。また、この例ではエントリがキャッシュされる時間の最大数もミリ秒で指定します。デフォルトのtime-to-live値は無限大です。この例では3600ミリ秒を指定しています。
<caching-system>
<name>caching-system</name>
<cache>
<name>localcache</name>
<max-size>64</max-size>
<eviction-policy>LFU</eviction-policy>
<time-to-live>3600</time-to-live>
</cache>
</caching-system>
次の構成ファイル設定は、write-behind要素をキャッシュの子要素として追加します。write-behind要素を指定すると、Oracle Event Processingでキャッシュ・エントリの作成または更新後に別のスレッドからキャッシュ・ストアを起動します。write-behindの子要素は次を示します。
バッキング・ストアに書込みを戻すためにストア・バッファからピック・アップされる更新の数(batch-size)。デフォルト値は100です。
ユーザー・スレッドがストア・バッファに書き込むために行う試行回数。ユーザー・スレッドとは、キャッシュ・エントリの作成または更新を行うスレッドです。ユーザー・スレッドによるストア・バッファへの書込み試行がすべて失敗した場合、同期的にストアを起動します(batch-write-attempts)。デフォルト値は1です。
ユーザー・スレッドがストア・バッファへの書込み試行を中断する前に待機する、ミリ秒単位の時間(buffer-write-timeout)。ストア・バッファへの書込み試行が失敗するのは、バッファが一杯の場合のみです。タイムアウト後は、以降のバッファへの書込み試行は、buffer-write-attemptsの値に基づいて行われます。デフォルト値は100です。
<caching-system> <name>caching-system-id</name> <cache> <name>cache-id</name> <max-size>100000</max-size> <eviction-policy>LRU</eviction-policy <time-to-live>3600</time-to-live> <write-behind> <buffer-size>200</buffer-size> <buffer-write-attempts>2</buffer-write-attempts> <buffer-write-timeout>200</buffer-write-timeout> </write-behind> </cache> </caching-system>
次の構成ファイル設定は、listeners子要素を追加し、キャッシュをリスニングするコンポーネントの動作を構成します。listener要素にはasynchronous属性があり、true (リスナーは非同期的に起動)またはfalse (リスナーは同期的に起動)のいずれかに設定できます。
work-manager-name子要素は、非同期的なリスナーの起動に使用されるワーク・マネージャを指定します。この値は、同期的起動が有効化されている場合は無視されます。キャッシュにワーク・マネージャが指定されている場合は、リスナーの呼出しについてのみこの値でオーバーライドされます。work-manager-name要素の値は、Oracle Event Processingサーバー構成ファイル(config.xml)のwork-manager設定のname要素と一致します。
<caching-system> <name>caching-system-id</name> <cache> <name>cache-id</name> <max-size>100000</max-size> <eviction-policy>LRU</eviction-policy <time-to-live>3600</time-to-live> <write-behind> <buffer-size>200</buffer-size> <buffer-write-attempts>2</buffer-write-attempts> <buffer-write-timeout>200</buffer-write-timeout> </write-behind> <listeners asynchronous="true"> <work-manager-name>cachingWM</work-manager-name> </listeners> </cache> </caching-system>
キャッシュを構成してネットワークを通過するイベントを受信できます。たとえば、キャッシュがチャネルをリスニングすることを指定するには、キャッシュへの参照を持つwlevs:listener要素とともにチャネルを構成します。
次の例では、チャネルが新しいイベントをキャッシュに送信するときに、イベントがキャッシュに挿入されます。チャネルでremove event(出力ウィンドウに存在する古いイベント)を送信する場合、そのイベントはキャッシュから削除されます。
<wlevs:caching-system id="caching-system-id"/>
<wlevs:cache id="cache-id" name="alternative-cache-name">
<wlevs:caching-system ref="caching-system-id"/>
</wlevs:cache>
<wlevs:channel id="tradeStream">
<wlevs:listener ref="cache-id"/>
</wlevs:channel>
次の項では、キャッシュのインデックスに使用するキーの指定に使用可能なオプションについて説明します。
明示的にキーを指定しない場合、イベントがキャッシュに挿入されるときにイベント・オブジェクトはキーおよび値の両方として提供されます。この場合、イベント・クラスには、キー・プロパティ値を考慮するequalsおよびhashcodeメソッドの有効な実装が含まれる必要があります。
次の例で示されるように、key-properties属性を使用してアセンブリ・ファイル内でキー・プロパティのプロパティ名を指定します。
<wlevs:cache id="myCache" key-properties="key-property-name">
<wlevs:caching-system ref="caching-system-id"/>
</wlevs:cache>
この場合、キャッシュに挿入されるすべてのイベントで実行時にこの名前のプロパティを保持することが要求され、それ以外の場合はOracle Event Processingで例外がスローされます。たとえば、キャッシュに挿入されているイベント型は次のように見えると仮定します。ここで、keyプロパティに注意します(関連するJavaソースのみが示されます)。
public class MyEvent {
private String key;
public MyEvent() {}
public MyEvent(String key) { this.key = key; }
public String getKey() { return key;}
public void setKey(String key) { this.key = key;}
}
アセンブリ・ファイルの対応する宣言は、次のようになります。
<wlevs:cache id="myCache" key-properties="key">
<wlevs:caching-system ref="caching-system-id"/>
</wlevs:cache>
メタデータ注釈com.bea.wlevs.ede.api.Keyを使用して、イベント型を実装するJavaクラスにおけるイベント・プロパティに注釈を付加できます。この注釈には、属性はありません。
メタデータ注釈を使用してキーを指定するには、次のステップを実行します。
wlevs:cache要素のkey-class属性を使用して、複数のプロパティがキーを形成する複合キーを指定できます。key-class属性値は、パブリック・フィールドがイベント・クラスのフィールドに一致するJavaBeanである必要があります。JavaBeanクラスは、java.lang.Object classからのequalsおよびhashCodeメソッドをオーバーライドする必要があります。照合はフィールド名に従って実行されます。次に例を示します。
<wlevs:cache id="myCache" key-class="key-class-name"> <wlevs:caching-system ref="caching-system-id"/> </wlevs:cache>
key-field1およびkey-field2で構成される複合キーを持つキャッシュに対しては、次の問合せを両方とも実行できます。
SELECT stream.field2, cache.key-field1 from stream[NOW], cache WHERE stream.field2=cache.key-field1 AND stream.field2=cache.key-field2 SELECT stream.field1, cache.key-field1 from stream[NOW], cache WHERE stream.field1=cache.key-field1
キャッシュをイベント・ソースとして構成できます。キャッシュをイベント・ソースとして使用するには、com.bea.wlevs.ede.api.StreamSinkインタフェースを実装する必要があります。
次に構成を示します。
<wlevs:cache id="cache-id" name="alternative-cache-name" caching-system="caching-system-id"> <wlevs:listener ref="cache-listener-id" /> </wlevs:cache>
キャッシュは、イベント処理ネットワーク内の別のコンポーネントがリスニングする対象のイベント・ソースとして構成可能です。リスニングするコンポーネントは、アダプタまたはBeanです。
キャッシュをリスニングするクラスは、次のようなイベントを受信する方法を提供するインタフェースを実装する必要があります。
Coherenceキャッシュをリスニングするクラスは、com.tangosol.util.MapListenerインタフェースを実装する必要があります。
Oracle Event Processingローカル・キャッシュをリスニングするクラスは、com.bea.cache.jcache.CacheListenerインタフェースを実装する必要があります。
<wlevs:caching-system id="caching-system-id"/> ... <wlevs:cache id="cache-id" name="alternative-cache-name"> <wlevs:caching-system ref="caching-system-id"/> <wlevs:cache-listener ref="cache-listener-id" /> </wlevs:cache> ... <bean id="cacheListenerId" class="com.bea.wlevs.example.provider.coherence"/>
例で、cacheListenerId Spring Beanは、キャッシュからのイベントをリスニングします。この場合、このコンポーネントを実装するユーザー定義クラスcom.bea.wlevs.example.MyCacheListenerは、Oracle Coherenceキャッシュをリスニングしています。com.tangosol.util.MapListenerを含む適切なOracle Coherence固有のJavaインタフェースを実装する必要があります。次の例は、この実装を示します。
package com.bea.wlevs.example.provider.coherence;
import com.tangosol.util.MapEvent;
import com.tangosol.util.MapListener;
public class LocalListener implements MapListener {
public static int deleted = 0;
public static int inserted = 0;
public static int updated = 0;
public void entryDeleted(MapEvent event) { deleted++; }
public void entryInserted(MapEvent event) { inserted++; }
public void entryUpdated(MapEvent event) { updated++; }
}
サードパーティのキャッシング・システムおよびキャッシュを使用するようにアプリケーションを構成できます。
サードパーティのキャッシング・システムおよびキャッシュの構成
EPNのキャッシュとデータベースを含む他のデータ・ソースのデータを交換できます。たとえば、アプリケーションの起動時にデータを使用してキャッシュをロードしたり、キャッシュとデータベース間の読取り/書込み関係を作成できます。
バッキング・ストアが読取り専用である場合を含めてキャッシュがデータを読み取るだけの場合、キャッシュ・ローダーを使用する必要があります。キャッシュでデータを読み取るまたは書き込む場合、キャッシュ・ストアを使用します。どちらの場合も、関係の作成には、特定の構成およびデータ・ソースとの通信方法を認識するJavaクラスが含まれます。
キャッシュ・ローダーを使用して、EPNのキャッシュに読取り専用データ・ソースからデータをロードできます。キャッシュ・ローダーはキャッシュにキャッシュ・オブジェクトをロードするJavaクラスです。適切なインタフェースを実装してキャッシュと通信するローダー・クラスを有効にするJavaクラスを書き込むことによって、キャッシュ・ローダーを作成します。wlevs:cache要素のwlevs:cache-loader子要素を使用してロード処理を行うBeanを指定することによって、キャッシュ・ローダーを構成します。
バッキング・ストアが読取り/書込みである場合は、かわりにキャッシュ・ストアを使用します(読取り/書込みデータ・ソースでのデータの交換を参照)。
キャッシュ・ローダーを作成する場合、次のようにインタフェースを実装します。
キャッシュ・データをOracle Coherenceキャッシュにロードするには、com.tangosol.net.cache.CacheLoaderを含む適切なOracle Coherence固有のJavaインタフェースを実装するクラスを作成します。例8-2を参照してください。
キャッシュ・データをOracle Event Processingローカル・キャッシュにロードするには、com.bea.cache.jcache.CacheLoaderインタフェースを実装するクラスを作成します。このインタフェースには、単一のオブジェクトのキャッシュへのロードをカスタマイズするためのloadメソッドが含まれます。Oracle Event Processingは、リクエストされたオブジェクトがキャッシュ内にないときにこのメソッドをコールします。このインタフェースには、キャッシュ全体のロードをカスタマイズするために実装するloadAllメソッドも含まれます。
例8-1では、localLoaderのBeanは、バッキング・ストアが読取り専用のときにイベントをOracle Coherenceキャッシュにロードします。
Coherenceキャッシュを使用していて構成ファイルでキャッシュ・ローダーを指定した場合、Coherenceキャッシュ構成ファイルで対応するクラス・ファクトリ・メソッド名も指定する必要があることに注意してください。キャッシュ・ローダーには、com.bea.wlevs.cache.coherence.configuration.SpringFactoryのgetLoaderメソッドを指定します。コードの例は、Oracle Coherenceキャッシュ・システムおよびキャッシュの構成を参照してください。
例8-1 キャッシュ・ローダー用Oracle CoherenceキャッシュのEPNアセンブリ・ファイル
<wlevs:caching-system id="caching-system-id"/> <wlevs:cache id="myCache" advertise="false"> <wlevs:caching-system ref="caching-system-id"/> <wlevs:cache-loader ref="localLoader"/> </wlevs:cache> <bean id="localLoader" class="com.bea.wlevs.example.provider.coherence.LocalLoader"/>
例8-2 Oracle CoherenceキャッシュのLocalLoaderの実装
package com.bea.wlevs.example.provider.coherence;
import java.util.Collection;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;
import com.bea.wlevs.example.provider.event.ProviderData;
import com.tangosol.net.cache.CacheLoader;
public class LocalLoader implements CacheLoader {
public static int loadCount = 0;
public static Set keys = new HashSet();
public LocalLoader() {
}
public Object load(Object key) {
loadCount++;
keys.add(key);
return new ProviderData((String) key);
}
public Map loadAll(Collection keys) {
Map result = new HashMap();
for (Object key : keys) {
result.put(key, load(key));
}
return result;
}
}
キャッシュ・ストアを使用すると、EPNのキャッシュで読取り/書込みデータ・ソースとデータを交換できます。キャッシュ・ストアはキャッシュとキャッシュ・オブジェクトを交換するJavaクラスです。適切なインタフェースを実装してデータソースと通信するために有効にするJavaクラスを書き込むことによって、キャッシュ・ストアを作成します。次に、wlevs:cache要素のwlevs:cache-store子要素を使用してデータソースと通信するBeanを指定することによって、キャッシュ・ストアをEPNに追加します。
バッキング・ストアが読取り専用である場合は、かわりにキャッシュ・ローダーを使用します(読取り専用データ・ソースからのキャッシュ・データのロードを参照)。
キャッシュ・ストアを作成する場合、次のようにインタフェースを実装します。
キャッシュ・データとOracle Coherenceキャッシュを交換するには、com.tangosol.net.cache.CacheStoreを含む適切なOracle Coherence固有のJavaインタフェースを実装するクラスを作成します。この例は、例8-4を参照してください。
キャッシュ・データとOracle Event Processingローカル・キャッシュを交換するには、com.bea.cache.jcache.CacheStoreインタフェースを実装するクラスを作成します。このインタフェースには、受け渡されたキーを使用してバッキング・ストアにデータを保存するstoreメソッドが含まれます。Oracle Event Processingは、データをキャッシュに挿入するときにこのメソッドをコールします。このインタフェースは、write-behind構成要素を使用してキャッシュに非同期書込みを構成した場合に、バッキング・ストアへのデータの一括保存に使用するstoreAllメソッドも含みます。
例8-3では、localStoreのBeanは、バッキング・ストアが読取り/書込みのときにイベントをキャッシュにロードします。
Spring構成ファイルでキャッシュ・ストアを指定した場合、Coherenceキャッシュ構成ファイルで対応するクラス・ファクトリ・メソッド名も指定する必要があることに注意してください。キャッシュ・ストアには、com.bea.wlevs.cache.coherence.configuration.SpringFactoryのgetStoreメソッドを指定します。コードの例は、Oracle Coherenceキャッシュ・システムおよびキャッシュの構成を参照してください。
例8-3 キャッシュ・ストア用Oracle CoherenceキャッシュのEPNアセンブリ・ファイル
<wlevs:caching-system id="caching-system-id"/> <wlevs:cache id="myCache" advertise="false"> <wlevs:caching-system ref="caching-system-id"/> <wlevs:cache-store ref="localStore"/> </wlevs:cache> <bean id="localStore" class="com.bea.wlevs.example.provider.coherence.LocalStore"/>
例8-4 Oracle CoherenceキャッシュのLocalStoreの実装
package com.bea.wlevs.example.provider.coherence;
import java.util.Collection;
import java.util.HashMap;
import java.util.Map;
import java.util.Set;
import com.bea.wlevs.example.provider.event.ProviderData;
import com.tangosol.net.cache.CacheStore;
public class LocalStore implements CacheStore {
public static int eraseCount = 0;
public static int storeCount = 0;
public static int loadCount = 0;
public void erase(Object key) {
eraseCount++;
}
public void eraseAll(Collection keys) {
for (Object key : keys) {
erase(key);
}
}
public void store(Object key, Object value) {
//
// Do the store operation here.
//
}
public void storeAll(Map entries) {
for (Map.Entry entry : (Set <Map.Entry>)entries.entrySet()) {
store(entry.getKey(), entry.getValue());
}
}
public Object load(Object key) {
loadCount++;
return new ProviderData((String) key);
}
public Map loadAll(Collection keys) {
Map result = new HashMap();
for (Object key : keys) {
result.put(key, load(key));
}
return result;
}
}
キャッシュを構成すると、Oracle Event Processingアプリケーションのいくつかのコンポーネントからキャッシュにアクセスできます。
この項では、その実行方法について説明します。
詳細は、次の項を参照してください。
アプリケーションをアセンブルおよびデプロイする前に、META-INF/MANIFEST.MFを編集して実装で必要になるパッケージをインポートします。たとえば、アプリケーションでキャッシュ・リスナー、キャッシュ・ローダーまたはキャッシュ・ストアを実装する場合は、Coherence APIを含むcom.tangosol.net.cacheパッケージを実装します。
Oracle Event Processingには、アプリケーションで特定のタスクを実行するために使用できるキャッシングAPIが用意されています。APIは、キャッシュへのアクセス、およびキャッシュ・ローダー、リスナー、ストアの作成に使用されるAPIを含むcom.bea.cache.jcacheパッケージ内にあります。ローダー機能、リスナー機能およびストア機能を使用するには、com.tangosol.netパッケージおよびcom.tangosol.net.cacheパッケージをインポートします。
EPNアセンブリ・ファイルおよびコンポーネント構成ファイルを使用して、キャッシング・システムおよびキャッシュを作成、構成および接続します。この場合、通常はアプリケーションでCacheおよびCachingSystemインタフェースを明示的には使用しません。これらは標準構成を上回る追加要件がある場合にのみ使用します。たとえば、サード・パーティ製キャッシュ・プロバイダとの統合を提供する場合は、CachingSystemインタフェースを使用する必要があります。キャッシュでjava.util.Mapインタフェース以外の操作を実行する必要がある場合は、Cacheインタフェースを使用できます。
Oracle Event Processingのローカル・キャッシュ用のキャッシュ・リスナー、キャッシュ・ローダー、またはキャッシュ・ストアを作成する場合、書き込むBeanはCacheListener、CacheLoader、またはCacheStoreインタフェースを実装する必要があります。
Oracle Coherenceキャッシュ用のキャッシュ・リスナー、キャッシュ・ローダー、またはキャッシュ・ストアを作成する場合、書き込むBeanは適切なOracle Coherenceインタフェースを実装する必要があります。
サードパーティのキャッシュ用のキャッシュ・リスナー、キャッシュ・ローダー、またはキャッシュ・ストアを作成する場合、書き込むBeanは適切なサードパーティのキャッシュ・インタフェースを実装する必要があります。
チャネルなどのイベント・ソースを参照するのとまったく同じ方法でOracle CQL文からキャッシュを参照できます。この機能によって、標準ストリーミング・データを個別ソースからのデータを使用して質を高めることができます。次の例のコードは、S1という標準チャネルからの取引イベントとstockCacheというキャッシュからの銘柄記号データを結合する、有効なOracle CQL問合せを示します。
Oracle CQL問合せにおいてキャッシュを使用するとき、次の制約を遵守する必要があります。
キャッシュに問い合せる場合は、常に[Now]ウィンドウに対して結合する必要があります。
これによって、問合せはキャッシュのスナップショットに対して実行されることが保証されます。他のウィンドウ・タイプに対して結合する場合、ウィンドウの有効期限が切れる前にキャッシュが変更されると、問合せは正しくなりません。
次の例は、キャッシュに対してRangeウィンドウを結合する無効なOracle CQL問合せを示します。このウィンドウの有効期限が切れる前にキャッシュが変更される場合、問合せは正しくなりません。その結果、この問合せはOracle Event Processingサーバー・エラー「外部リレーションはs[now]と結合している必要があります」を表示します。
SELECT trade.symbol, trade.price, trade.numberOfShares, company.name FROM TradeStream [Range 8 hours] as trade, CompanyCache as company WHERE trade.symbol = company.id
Oracle CQL問合せでキャッシュからのデータを使用する場合、チャネルの場合と同様に、データが押し出されるのではなく、Oracle Event Processingがデータを抽出します。つまり、問合せはチャネルがtradeイベントを問合せに押し出すときのみ実行されます。キャッシュ内のストック・シンボル・データは問合せを実行せず、必要に応じて問合せによって抽出されるのみです。
キャッシュ・キーに基づいて参照を行うには、必要なキー・プロパティを指定する必要があります。
スキーマ(id、group、value)を持つ(キャッシュ・キーはid)、2つのストリームSおよびCを検討してください。有効な問合せは次のとおりです。
select count(*) as n from S [now], C where S.id = C.id
結合はキャッシュのキーを参照することでのみ実行される必要があります。
ビューでキャッシュは使用できません。かわりに、結合を使用します。
キャッシュ・データ・ソースを結合するOracle CQL文のFROM句では、1つのチャネル・ソースのみが発生できます。
キャッシュがプロセッサ・ソースである場合は、キャッシュをEPNのチャネルに直接接続します。
キャッシュがプロセッサ・シンクである場合は、プロセッサに直接接続できます。
Oracle CQL文からキャッシュへのアクセス
この手順では、キャッシング・システムおよびキャッシュを構成済と仮定します。詳細は、次を参照してください。
まだ構成済でない場合、キャッシュ・データに対応するイベント型を作成し、イベント・リポジトリに登録します。
イベントおよびイベント・タイプを参照してください。
キャッシュ内のデータからキー・プロパティを指定します。
EPNアセンブリ・ファイルで、キャッシュの構成を更新してその値のイベント型を宣言します。wlevs:cache要素のvalue-type属性を使用します。次に例を示します。
<wlevs:caching-system id="caching-system-id"/>
...
<wlevs:cache id="cache-id"
name="alternative-cache-name"
value-type="CompanyEvent">
<wlevs:caching-system ref="caching-system-id"/>
</wlevs:cache>
value-type属性は、キャッシュに含まれる値のタイプを指定します。これは、イベント型リポジトリで有効なタイプ名にする必要があります。
この属性は、Oracle CQL問合せでキャッシュが参照される場合に必要です。これは、問合せプロセッサはキャッシュ内のイベントのタイプを意識している必要があるからです。
EPNアセンブル・ファイルで、キャッシュを参照するOracle CQL問合せを実行するプロセッサの構成を更新します。
キャッシュがプロセッサ・ソースである場合は、図8-1に示すように、キャッシュをEPNのプロセッサに直接接続します。
キャッシュを参照するwlevs:processor要素のwlevs:cache-source子要素を更新します。次に例を示します。
<wlevs:channel id="S1"/>
<wlevs:processor id="cacheProcessor">
<wlevs:source ref="S1">
<wlevs:cache-source ref="cache-id">
</wlevs:processor>
この例では、プロセッサは従来どおりにS1チャネルから押し出されるデータを持ちます。ただし、プロセッサで実行するOracle CQL問合せは、cache-idキャッシュからデータを抽出することもできます。問合せプロセッサが、FROM句のイベント型をCompanyEventなどのキャッシュによって提供されるイベント型に一致するとき、プロセッサはキャッシュからそのイベント型のインスタンスを抽出します。
キャッシュがプロセッサ・シンクである場合は、図8-2に示すように、EPNのチャネルを使用してプロセッサをキャッシュに接続します(つまり、プロセッサとキャッシュ・シンクの間にチャネルが必要です)。
この場合、アプリケーション・アセンブリ・ファイルは次のようになります。
<wlevs:channel id="channel1" event-type="StockTick">
<wlevs:listener ref="processor" />
</wlevs:channel>
<wlevs:processor id="processor">
<wlevs:listener ref="channel2" />
</wlevs:processor>
<wlevs:channel id="channel2" event-type="StockTick">
<wlevs:listener ref="cache-id" />
</wlevs:channel>
SELECT S1.symbol, S1.lastPrice, stockCache.description FROM S1 [Now], stockCache WHERE S1.symbol = stockCache.symbol
アダプタも、他のBeanを参照するための標準Springメカニズムを使用してキャッシュに組み込むことができます。キャッシュBeanは、java.util.Mapインタフェースを実装します。このインタフェースは、組み込まれたキャッシュにアクセスするためにアダプタが使用します。
まず、EPNアセンブリ・ファイル内のアダプタの構成は、次の例で示すように、wlevs:instance-property子要素を使用して更新する必要があります。
<wlevs:caching-system id="caching-system-id"/>
...
<wlevs:cache id="cache-id" name="alternative-cache-name">
<wlevs:caching-system ref="caching-system-id"/>
</wlevs:cache>
...
<wlevs:adapter id="myAdapter" provider="myProvider">
<wlevs:instance-property name="map" ref="cache-id"/>
</wlevs:adapter>
この例では、wlevs:instance-propertyのref属性はwlevs:cache要素のid値を参照します。Oracle Event Processingは、自動的にアダプタにキャッシュ(java.util.Mapとして実装)を組み込みます。
アダプタのJavaソースでは、アダプタによるキャッシュ処理の機能を実装するコードを持つsetMap (Map)メソッドを追加します。
package com.bea.wlevs.example;
…
import java.util.Map;
public class MyAdapter implements Runnable, Adapter, EventSource, SuspendableBean {
...
public void setMap (Map map) {...}
}
EPNアセンブリ・ファイルで標準のSpring Beanとして構成されたビジネスPOJOでは、他のBeanを参照するための標準のSpringメカニズムを使用して、キャッシュを挿入することができます。この方法によって、POJOではキャッシュを表示および操作できます。キャッシュBeanでは、ビジネスPOJOが挿入されたキャッシュにアクセスするために使用するjava.util.Mapインタフェースが実装されます。キャッシュBeanではjava.util.Mapのベンダー固有のサブインタフェースを実装することもできますが、移植性のためにMapを実装することをお薦めします。
まず、EPNアセンブリ・ファイル内のビジネスPOJOの構成は、FX例(『Oracle Event Processingスタート・ガイド』の外国為替(FX)の例に関する項を参照)のOutput Beanに基づいた次の例で示されるように、property子要素を使用して更新する必要があります。
<wlevs:caching-system id="caching-system-id"/>
...
<wlevs:cache id="cache-id" name="alternative-cache-name">
<wlevs:caching-system ref="caching-system-id"/>
</wlevs:cache>
...
<bean class="com.bea.wlevs.example.helloworld.HelloWorldBean">
<property name="map" ref="cache-id"/>
</bean>
この例では、property要素のref属性はwlevs:cache要素のid値を参照します。Oracle Event Processingは、自動的にビジネスPOJO Beanにキャッシュ(java.util.Mapとして実装)を組み込みます。
ビジネスPOJO BeanのJavaソースでは、POJOによるキャッシュ処理の機能を実装するコードを持つsetMap (Map)メソッドを追加します。
package com.bea.wlevs.example.helloworld;
…
import java.util.Map;
public class HelloWorldBean implements EventSink {
...
public void setMap (Map map) {...}
}
標準イベント・ストリームに加えて、Oracle CQLルールはユーザー定義関数のメンバー・メソッドも起動できます。
これらのユーザー定義関数は標準Javaクラスとして実装され、次の例で示すように、Oracle CQLプロセッサのコンポーネント構成ファイル内で宣言されます。
<bean id="orderFunction" class="orderFunction-impl-class"/>
関連するOracle CQLルールが実行されるプロセッサは、ref属性を持つSpring Beanを参照し、wlevs:function子要素を使用してユーザー定義関数に組み込まれる必要があります。
<wlevs:processor id= "tradeProcessor">
<wlevs:function ref="orderFunction"/>
</wlevs:processor>
また、wlevs:function要素でBeanクラスを指定できます。
<wlevs:processor id="testProcessor"> <wlevs:listener ref="providerCache"/> <wlevs:listener ref="outputCache"/> <wlevs:cache-source ref="testCache"/> <wlevs:function function-name="mymod" exec-method="execute" /> <bean class="com.bea.wlevs.example.function.MyMod"/> </wlevs:function> </wlevs:processor>
次のOracle CQLルールは、tradeProcessorプロセッサに構成されると仮定され、orderFunctionユーザー定義関数のexistsOrderメソッドを起動する方法を示します。
INSERT INTO InstitutionalOrder
SELECT er.orderKey AS key, er.symbol AS symbol, er.shares as cumulativeShares
FROM ExecutionRequest er [Range 8 hours]
WHERE NOT orderFunction.existsOrder(er.orderKey)
別のBeanを参照する標準Springメカニズムを使用してキャッシュを持つ関数を組み込むことによって、キャッシュにアクセスするためのユーザー定義関数を構成することもできます。キャッシュBeanは、java.util.Mapインタフェースを実装します。このインタフェースは、組み込まれたキャッシュにアクセスするためにユーザー定義関数が使用します。
まず、EPNアセンブリ・ファイル内のユーザー定義関数の構成は、次の例で示すように、wlevs:property子要素を使用して更新する必要があります。
<wlevs:caching-system id="caching-system-id"/>
...
<wlevs:cache id="cache-id" name="alternative-cache-name">
<wlevs:caching-system ref="caching-system-id"/>
</wlevs:cache>
...
<bean id="orderFunction" class="orderFunction-impl-class">
<wlevs:property name="cache" ref="cache-id"/>
</bean>
この例では、wlevs:property要素のref属性はwlevs:cache要素のid値を参照します。Oracle Event Processingは、自動的にユーザー定義関数にキャッシュ(java.util.Mapとして実装)を組み込みます。
ユーザー定義関数のJavaソースでは、関数によるキャッシュ処理の機能を実装するコードを持つsetMap (Map)メソッドを追加します。
package com.bea.wlevs.example;
…
import java.util.Map;
public class OrderFunction {
...
public void setMap (Map map) {...}
}
ユーザー定義関数の詳細は、『Oracle Event Processing Oracle CQL言語リファレンス』のユーザー定義に関する項を参照してください。
実行時、Oracle Event Processingがキャッシング・システムと定義するキャッシュにデプロイするMBeansおよびJMXを使用して、キャッシュにプログラムからアクセスできます。詳細は、『Oracle Event Processingの管理』のJMXに関する項を参照してください。
JMXを使用してキャッシュ・システムまたはキャッシュにアクセスする最も単純かつエラーの少ない方法は、Oracle Event Processing Visualizerを使用することです。詳細は、『Oracle Event Processingビジュアライザの使用』のJMXの管理に関する項を参照してください。
JMXを使用してキャッシュ・システムまたはキャッシュにアクセスする最も単純かつエラーの少ない方法は、Oracle Event Processing Visualizerを使用することです(Oracle Event Processing Visualizerを使用したJMXによるキャッシュへのアクセス方法を参照)。あるいは、書き込まれるJavaコードを使用したJMXによるキャッシング・システムまたはキャッシュにアクセスできます。
Oracle Event Processingは、アプリケーションがステージとして使用する各キャッシュのStageMBeanを作成します。このMBeanのTypeはStageです。
Javaを使用したJMXでキャッシュにアクセスするには:
