15 キャッシュ・イベントの使用(C++)

マップ・イベント・リスナーを使用して、ObservableMapインタフェースを実装するCoherence内の任意のクラスからキャッシュ・イベントおよびイベントを受信できます。

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

• マップ・イベントの概要(C++)
  Coherenceには、MapListenerインタフェースが用意されており、これによりCoherenceキャッシュのデータが追加、変更または削除されたときにアプリケーション・ロジックでイベントを受信できます。
• イベントをサポートするキャッシュとクラス
  すべてのCoherenceキャッシュには、アプリケーションがキャッシュ・イベントを受信できるようにするObservableMapインタフェースが実装されています。
• すべてのイベントのサインアップ
  イベントにサインアップするには、MapListenerインタフェースを実装するオブジェクトをObservableMapのaddMapListenerメソッドに渡します。
• 多重化マップ・リスナーの使用
  MultiplexingMapListenerクラスは、すべてのイベントを1つのメソッドにルーティングして処理します。
• キャッシュ用のMapListenerの構成
  キャッシュ構成内の<listener>要素を使用して、リスナーをキャッシュに登録できます。
• 特定のIDのイベントのサインアップ
  特定のID (キー)に対して発生したイベントにサインアップできます。
• イベントのフィルタリング
  フィルタを使用して、特定のイベントをリスニングできます。
• Liteイベントの使用
  アプリケーションで古い値と新しい値をイベントに含める必要がない場合は、Liteイベントを使用してリソースを節約できます。
• 問合せのリスニング
  Coherenceキャッシュは、任意の基準による問合せをサポートします。
• 統合イベントの使用
  アプリケーションは、キャッシュ内の操作から発生した統合イベントをリスニングできます。
• バッキング・マップ・イベントの使用
  いくつかの高度なユース・ケースでは、サービスの背後にあるマップのイベントをリスニングできます。
• 同期イベント・リスナーの使用
  MapListener実装では、SynchronousListenerマーカー・インタフェースを使用して、クラスタ化されたシステムのローカル・ビューがシングル・スレッドであるかのように、キャッシュAPIの操作およびイベントが順序付けされることを保証できます。

マップ・イベントの概要(C++)

イベント・モデルはEventListenerインタフェースから成り、すべてのリスナーでこれを拡張する必要があります。Coherenceには、MapListenerインタフェースが用意されており、これによりCoherenceキャッシュのデータが追加、変更または削除されたときにアプリケーション・ロジックでイベントを受信できます。

MapListenerインタフェースを実装するアプリケーション・オブジェクトは、アプリケーションのMapListener実装のインスタンスをaddMapListener()メソッドに渡すことで、ObservableMapインタフェースを実装する任意のCoherenceキャッシュまたはクラスからイベントにサインアップできます。

MapListenerに渡されるMapEventオブジェクトは、イベントの発生元になったソース(ObservableMap)、イベントに関連付けられたID(キー)、そのIDに対するアクション(挿入、更新、削除)、古い値と新しい値など、発生したイベントについて必要なすべての情報を保持します。

イベントをサポートするキャッシュとクラス

すべてのCoherenceキャッシュには、アプリケーションがキャッシュ・イベントを受信できるようにするObservableMapインタフェースが実装されています。すべてのキャッシュは、そのキャッシュが、ローカル、パーティション、ニア、レプリケートであるかどうか、また、リードスルー、ライトスルー、ライトビハインド、オーバーフロー、ディスク記憶域などを使用しているかどうかに関係なく、イベントを受信できます。

ノート:

キャッシュ・トポロジやサーバーの数に関係なく、また他のサーバーが変更している場合でも、これらのイベントはアプリケーションのリスナーに配信されます。

Coherenceキャッシュ(Coherenceキャッシュ・ファクトリを介して取得されたオブジェクト)に加えて、次のように、Coherenceでサポートされる他のいくつかのクラスでもObservableMapインタフェースが実装されます。

• ObservableHashMap

• LocalCache

• OverflowMap

• NearCache

• ReadWriteBackingMap

• AbstractSerializationCache、SerializationCacheおよびSerializationPagedCache

• WrapperObservableMap、WrapperConcurrentMapおよびWrapperNamedCache

公開されている実装クラスの一覧は、Coherence APIの「ObservableMap」を参照してください。

すべてのイベントのサインアップ

イベントにサインアップするには、MapListenerインタフェースを実装するオブジェクトをObservableMapのaddMapListenerメソッドに渡します。

たとえば:

virtual void addKeyListener(MapListener::Handle hListener, Object::View vKey, bool fLite) = 0;
virtual void removeKeyListener(MapListener::Handle hListener, Object::View vKey) = 0;
virtual void addFilterListener(MapListener::Handle hListener, Filter::View vFilter = NULL, bool fLite = false) = 0;
virtual void removeFilterListener(MapListener::Handle hListener, Filter::View vFilter = NULL) = 0;

MapListener実装のサンプルを作成してみましょう。

#include "coherence/util/MapEvent.hpp"
#include "coherence/util/MapListener.hpp"

#include <iostream>

using coherence::util::MapEvent;
using coherence::util::MapListener;
using namespace std;

/**
* A MapListener implementation that prints each event as it receives
* them.
*/
class EventPrinter 
    : public class_spec<EventPrinter,
        extends<Object>,
        implements<MapListener> >
    {    
    friend class factory<EventPrinter>;

    public:
        virtual void entryInserted(MapEventView vEvent)
            {
            cout << vEvent << endl;
            }

        virtual void entryUpdated(MapEventView vEvent)
            {
            cout << vEvent << endl;
            }

        virtual void entryDeleted(MapEventView vEvent)
            {
            cout << vEvent << endl;
            }
    };

この実装を使用すると、任意のキャッシュからすべてのイベントを簡単に出力できます(すべてのキャッシュでObservableMapインタフェースが実装されているため)。

NamedCache::Handle hCache;
...
hCache->addFilterListener(EventPrinter::create());

ただし、後でリスナーを削除できるように、リスナーへの参照を保持する必要があります。

MapListener::Handle hListener = EventPrinter::create();
hCache->addFilterListener(hListener);
m_hListener = hListener; // store the listener in a member field

後でリスナーを削除するには、次のように処理します。

MapListener::Handle hListener = m_hListener;
if (hListener != NULL)
    {
    hCache->removeFilterListener(hListener);
    m_hListener = NULL; // clean up the listener field
    }

ObservableMapインタフェースの各add*Listenerメソッドには、対応するremove*Listenerメソッドがあります。リスナーを削除するには、そのリスナーの追加に使用したadd*Listenerメソッドに対応するremove*Listenerメソッドを使用します。

多重化マップ・リスナーの使用

MultiplexingMapListenerクラスは、すべてのイベントを1つのメソッドにルーティングして処理します。次の例は、単純なEventPrinterクラスを示しています。

#include "coherence/util/MultiplexingMapListener.hpp"

#include <iostream>

using coherence::util::MultiplexingMapListener;


class EventPrinter 
    : public class_spec<EventPrinter,
        extends<MultiplexingMapListener> >
    {
    public:
        virtual void onMapEvent(MapEventView vEvent)
            {
            std::cout << vEvent << std::endl;
            }
    };

キャッシュ用のMapListenerの構成

キャッシュ構成内の<listener>要素を使用して、リスナーをキャッシュに登録できます。構成されている場合、Coherenceでキャッシュを構成するときにリスナーが自動的に追加されます。リスナーを構成内に登録すると、特定のキャッシュが常にリスナーを必要とする場合に便利です。

特定のIDのイベントのサインアップ

特定のID(キー)に対して発生したイベントにサインアップできます。次のコードでは、Integerキー5に対して発生したすべてのイベントが出力されます。

hCache->addKeyListener(EventPrinter::create(), Integer32::create(5), false);

次のコードでは、Integerキー5が挿入または更新された場合にのみイベントがトリガーされます。

for (int32_t i = 0; i < 10; ++i)
    {
    Integer32::View vKey   = Integer32::create(i);
    Integer32::View vValue = vKey;
    hCache->put(vKey, vValue);
    }

イベントのフィルタリング

フィルタを使用して、特定のイベントをリスニングできます。次の例では、削除イベントの受信のみを許可するフィルタを指定したリスナーがキャッシュに追加されます。

// Filters used with partitioned caches must implement coherence::io::pof::PortableObject

#include "coherence/io/pof/PofReader.hpp"
#include "coherence/io/pof/PofWriter.hpp"
#include "coherence/io/pof/PortableObject.hpp"
#include "coherence/util/Filter.hpp"
#include "coherence/util/MapEvent.hpp"

using coherence::io::pof::PofReader;
using coherence::io::pof::PofWriter;
using coherence::io::pof::PortableObject;
using coherence::util::Filter;
using coherence::util::MapEvent;

class DeletedFilter
    : public class_spec<DeletedFilter,
        extends<Object>,
        implements<Filter, PortableObject> >
    {
    public:
        // Filter interface        virtual bool evaluate(Object::View v) const
            {
            MapEvent::View vEvt = cast<MapEvent::View>(v);
            return MapEvent::entry_deleted == vEvt->getId();
            }

        // PortableObject interface        virtual void readExternal(PofReader::Handle hIn)
            {
            }

        virtual void writeExternal(PofWriter::Handle hOut) const
            {
            }
    };

hCache->addFilterListener(EventPrinter::create(), DeletedFilter::create(), false);

たとえば、次の一連のコールが実行されたとします。

cache::put(String::create("hello"), String::create("world"));
cache::put(String::create("hello"), String::create("again"));
cache::remove(String::create("hello"));

次のような結果になります。

CacheEvent{LocalCache deleted: key=hello, value=again}

問合せのリスニングを参照してください。

イベントのフィルタリングとキャッシュされたデータのフィルタリングの比較

問合せのFilterを作成する場合、Filterのevaluateメソッドに渡すオブジェクトはキャッシュからの値になります。または、FilterでEntryFilterインタフェースを実装する場合は、キャッシュからのMap::Entry全体になります。MapListenerのイベントをフィルタリングするFilterを作成する場合、Filterのevaluateメソッドに渡すオブジェクトは常にMapEvent型になります。

Liteイベントの使用

アプリケーションで古い値と新しい値をイベントに含める必要がない場合は、Liteイベントを使用してリソースを節約できます。デフォルトでは、イベントの一部として古い値と新しい値が両方とも提供されます。次に例を示します。

MapListener::Handle hListener = EventPrinter::create();
// add listener with the default"lite" value of falsehCache->addFilterListener(hListener);

// insert a 1KB value
String::View vKey = String::create("test");
hCache->put(vKey, Array<octet_t>::create(1024));

// update with a 2KB value
hCache->put(vKey, Array<octet_t>::create(2048));

// remove the value
hCache->remove(vKey);

このコードを実行すると、挿入イベントには新しい1KBの値、更新イベントには古い1KBの値と新しい2KBの値、削除イベントには削除された2KBの値がそれぞれ保持されます。

リスナーの追加時にLiteイベントをリクエストするには、追加のブール・パラメータfLiteを使用するaddFilterListenerメソッドとaddKeyListenerメソッドのいずれかを使用できます。前述の例の場合、変更は次の箇所のみです。

cache->addFilterListener(hListener, (Filter::View) NULL, true);

ノート:

Liteイベントの古い値と新しい値はNULLにできます。ただし、Liteイベントをリクエストしても、イベントの生成と配信に余分な負荷がかからない場合は、古い値と新しい値が含まれることがあります。つまり、MapListener受信Liteイベントを要求することは、MapListenerがイベントについて古い値と新しい値の情報を必要としないことを単純にシステムに通知することになります。

問合せのリスニング

Coherenceキャッシュはどの基準による問合せもサポートしています。アプリケーションでキャッシュからのデータを問い合せる場合、その結果はID (keySet)のセットまたはID/値のペア(entrySet)のセットとしてのポイント・イン・タイム・スナップショットになります。結果セットの内容を判断するメカニズムは、フィルタリングと呼ばれ、これによりアプリケーション開発者は即時利用可能なフィルタ(equals、less-than、like、betweenなど)の豊富なセットを使用してどのような複雑な問合せをも構築でき、また独自のカスタム・フィルタ(XPathなど)を提供できます。

キャッシュの問合せに使用するフィルタと同じフィルタを、キャッシュからのイベントのリスニングに使用します。たとえば、トレーディング・システムにおいて、特定のトレーダーの未決済Orderオブジェクトをすべて問い合せることができます。

ノート:

次の例では、coherence::util::extractor::ReflectionExtractorクラスを使用しています。C++クライアントではリフレクションがサポートされませんが、クラスタ内で実行される問合せにはReflectionExtractorを使用できます。この場合、ReflectionExtractorは、問合せを実行するクラスタに、必要な抽出情報を渡すのみです。ReflectionExtractorがクライアントでデータを抽出することになる場合(値をローカルにキャッシュするときのContinuousQueryCacheなど)は、ReflectionExtractorの使用はサポートされません。この場合は、カスタム・エクストラクタを用意する必要があります。

NamedCache::Handle hMapTrades = ...
Filter::Handle hFilter = AndFilter::create(
        EqualsFilter::create(ReflectionExtractor::create("getTrader"), vTraderId),
        EqualsFilter::create(ReflectionExtractor::create("getStatus"), Status::OPEN));
Set::View vSetOpenTrades = hMapTrades->entrySet(hFilter);

そのトレーダーが新しいポジションを建てた、そのトレーダーがポジションを決済した、トレーダー間でポジションを付け替えた、という情報を受信するために、アプリケーションで同じフィルタを使用できます。

// receive events for all trade IDs that this trader is interested in
hMapTrades->addFilterListener(hListener, MapEventFilter::create(hFilter), true);

MapEventFilterは、問合せフィルタをイベント・フィルタに変換します。

ノート:

イベントのフィルタリングとキャッシュされたデータのフィルタリングの比較: 問合せのFilterを作成する場合、Filterのevaluateメソッドに渡すオブジェクトはキャッシュからの値になります。または、FilterでEntryFilterインタフェースを実装する場合は、キャッシュからのMap::Entry全体になります。MapListenerのイベントをフィルタリングするFilterを作成する場合、Filterのevaluateメソッドに渡すオブジェクトは常にMapEvent型になります。

MapEventFilterは、問合せに使用するFilterをMapListenerのイベントのフィルタリングに使用するFilterに変換します。つまり、MapEventFilterはキャッシュを問い合せるFilterから作成されますが、作成されたMapEventFilterは、問合せFilterが期待するオブジェクトに変換することによってMapEventオブジェクトを評価するフィルタになります。

MapEventFilterには、いくつもの非常に強力なオプションがあります。これにより、アプリケーション・リスナーは特に関心のあるイベントのみを受信できます。スケーラビリティとパフォーマンスの観点からさらに重要なこととして、必要なイベントのみがネットワーク上で通信されるため、それらの特定イベントを対象とするサーバーとクライアントにのみ配信されます。たとえば:

// receive all events for all trades that this trader is interested in
int32_t nMask = MapEventFilter::e_all;
hMapTrades->addFilterListener(hListener, MapEventFilter::create(nMask, hFilter), true);

// receive events for all this trader's trades that are closed or
// re-assigned to a different trader
nMask = MapEventFilter::e_updated_left | MapEventFilter::e_deleted;
hMapTrades->addFilterListener(hListener, MapEventFilter::create(nMask, hFilter), true);

// receive events for all trades as they are assigned to this trader
nMask = MapEventFilter::e_inserted | MapEventFilter::e_updated_entered;
hMapTrades->addFilterListener(hListener, MapEventFilter::create(nMask, hFilter), true);

// receive events only for new trades assigned to this trader
nMask = MapEventFilter::e_inserted;
hMapTrades->addFilterListener(hListener, MapEventFilter::create(nMask, hFilter), true);

統合イベントの使用

アプリケーションは、キャッシュ内の操作から発生した統合イベントをリスニングできます。統合イベントはクライアント・イベントとは異なります。

イベントは通常、キャッシュで発生している変更を反映します。たとえば、あるサーバーがキャッシュ内の1つのエントリを変更し、別のサーバーがキャッシュに複数のアイテムを追加し、3つ目のサーバーが同じキャッシュから1つのアイテムを削除している間に、クラスタ内の各サーバーで50個のスレッドが同じキャッシュのデータにアクセスすることがあります。すべての変更操作によりイベントが生成されますが、それをクラスタ内の任意のサーバーで受信するように選択できます。これらの操作をクライアント・アクションと呼び、そのイベントは、クライアントにディスパッチされた状態にあります(この場合のクライアントは実際にはサーバー)。これはCoherenceクラスタなどの、真のpeer-to-peerアーキテクチャでは自然の概念です。いずれのピアもクライアントでありサーバーでもあるので、他のピアからサービスを受け、他のピアにサービスを提供します。一般的なJava Enterpriseアプリケーションでは、ピアはアプリケーションのコンテナとして動作するアプリケーション・サーバー・インスタンスで、クライアントはキャッシュに対して直接アクセスおよび変更を行い、キャッシュからのイベントをリスニングするアプリケーションの一部です。

イベントの中には、キャッシュ内部で起動されるものもあります。多くの例がありますが、最も一般的なケースは次のとおりです。

• キャッシュのエントリが自動失効した場合

• キャッシュの最大サイズに達したため、エントリがキャッシュから削除された場合

• リードスルー操作の結果としてエントリが透過的にキャッシュに追加された場合

• リードアヘッドまたはリフレッシュアヘッド操作の結果としてキャッシュ内のエントリが透過的に更新された場合

これらは変更の各例ですが、変更とはキャッシュ内部での自然な(通常は自動的な)各操作を表します。これらのイベントを統合イベントと呼びます。

必要に応じて、アプリケーションはイベントに問い合せて、それがクライアントに起因するイベントか、または統合イベントかを区別できます。この情報は、CacheEventというMapEventのサブクラスで保持されます。前述のEventPrinterの例を使用して、統合イベントのみを出力できます。

class EventPrinter
    : public class_spec<EventPrinter,
        extends<MultiplexingMapListener> >
    {
    friend class factory<EventPrinter>;

    public:
        void onMapEvent(MapEvent::View vEvt)
            {
            if (instanceof<CacheEvent::View>(vEvt) &&
                (cast<CacheEvent::View>(vEvt)->isSynthetic()))
                {
                std::cout << vEvt;
                }
            }
    };

この機能の詳細は、CacheEventのAPIドキュメントを参照してください。

バッキング・マップ・イベントの使用

いくつかの高度なユース・ケースでは、サービスの背後にあるマップのイベントをリスニングできます。分散環境でのレプリケーション、パーティション、およびその他のデータ管理方式は、いずれも分散サービスです。実際にサービスのデータを管理するデータ構造は、バッキング・マップと呼ばれます。

バッキング・マップは構成可能です。特定のキャッシュのすべてのデータを、ヒープのオブジェクト形式で保持する必要がある場合、無制限で失効期限のないLocalCache(統計が不要な場合はSafeHashMap)を使用します。メモリーに保持するアイテムの数が少ない場合は、LocalCacheを使用します。データを必要に応じてデータベースから読み取る場合は、ReadWriteBackingMapを使用します(アプリケーションのDAO実装を介した読取りおよび書込み方法を指定します)。ReadWriteBackingMapに、SafeHashMapやLocalCacheなどのバッキング・マップを提供してデータを保存します。

バッキング・マップの中には観測可能なものがあります。これらのバッキング・マップから取得されたイベントは、一般的にはアプリケーションに直接関係がありません。かわりに、Coherenceではそれらを(Coherenceで)実行する必要があるアクションに変換してデータを同期的に保持し正しくバックアップを行い、またアプリケーション・リスナーでのリクエストに応じてそれをクラスタ全体に配信されるクラスタ化されたイベントに変換します。たとえば、パーティション・キャッシュにバッキング・マップとしてLocalCacheがあり、ローカル・キャッシュでエントリが失効した場合、そのイベントによってエントリのすべてのバックアップ・コピーが失効します。さらに、リスナーがパーティション・キャッシュに登録されている場合、イベントがイベント・フィルタに適合すると、そのイベントはそのリスナーが登録されているサーバーのリスナーに配信されます。

応用例の中には、データが保持されているサーバー上のイベントをアプリケーションで処理する必要があるものや、それを実際にデータを管理している構造(バッキング・マップ)で行う必要があるものがあります。そのような場合、バッキング・マップが観測可能マップであれば、リスナーをバッキング・マップに構成したり、プログラムによってリスナーをバッキング・マップに追加することができます。(バッキング・マップが監視可能でない場合、WrapperObservableMapでラップすることにより監視可能になります。)

同期イベント・リスナーの使用

イベントの中には、アプリケーション・リスナーがイベントを生成するキャッシュ・サービスを妨害しないように、非同期に配信されるものがあります。まれなケースですが、非同期配信では進行している処理の結果に比べてイベントの順序があいまいになる場合があります。

MapListener実装では、SynchronousListenerマーカー・インタフェースを使用して、クラスタ化されたシステムのローカル・ビューがシングル・スレッドであるかのように、キャッシュAPIの操作およびイベントが順序付けされることを保証できます。

Coherence自体で同期リスナーを使用する例にニア・キャッシュがあります。これにより、ローカルでキャッシュされたデータを、イベントを使用して無効化できます。