第10章 通知/リスナー API

キャッシュイベントが発生するたびに通知がリスナーに送信されます。リスナーは、@Listener アノテーションが付いた簡単な POJO です。Listenable は、実装がアタッチするリスナーを持つことを意味するインターフェースです。各リスナーは Listenable で定義されたメソッドを使用して登録されます。

キャッシュレベルまたはキャッシュマネージャーレベルの通知を受信するため、リスナーはキャッシュおよびキャッシュマネージャーの両方にアタッチすることが可能です。

Red Hat JBoss Data Grid では、キャッシュレベルのイベントはキャッシュごとに発生します。キャッシュレベルイベントの例には、関係するキャッシュで登録されたリスナーへの通知を引き起こすエントリーの追加、削除、および変更などが含まれます。

Red Hat JBoss Data Grid のキャッシュマネージャーレベルで発生するイベントの例は次のとおりです。

キャッシュマネージャーレベルのイベントはグローバルに位置し、クラスター全体で使用されますが、単一のキャッシュマネージャーによって作成されたキャッシュ内のイベントに制限されます。

最初の 2 つのイベントである CacheStarted および CacheStopped は大変似ています。以下の例は、開始または停止されたキャッシュの名前を出力します。

@CacheStarted
public void cacheStarted(CacheStartedEvent event){
    // Print the name of the Cache that started
    log.info("Cache Started: " + event.getCacheName());
}

@CacheStopped
public void cacheStopped(CacheStoppedEvent event){
    // Print the name of the Cache that stopped
    log.info("Cache Stopped: " + event.getCacheName());
}

ViewChangedEvent または MergeEvent を受信するとき、新旧メンバーのリストはイベントが生成されたノードから送信されることに注意してください。例として、以下のシナリオについて考えてみましょう。

よって、ノードのクラスターへの参加またはクラスターからの離脱を判断するために積集合が使用されることがあります。以下のように getOldMembers() と getNewMembers() を使用すると、クラスターに参加したノードや離脱したノードのセットを判断できます。

@ViewChanged
public void viewChanged(ViewChangedEvent event){
    HashSet<Address> oldMembers = new HashSet(event.getOldMembers());
    HashSet<Address> newMembers = new HashSet(event.getNewMembers());
    HashSet<Address> oldCopy = (HashSet<Address>)oldMembers.clone();

    // Remove all new nodes from the old view.
    // The resulting set indicates nodes that have left the cluster.
    oldCopy.removeAll(newMembers);
    if(oldCopy.size() > 0){
        for (Address oldAdd : oldCopy){
            log.info("Node left:" + oldAdd.toString());
        }
    }

    // Remove all old nodes from the new view.
    // The resulting set indicates nodes that have joined the cluster.
    newMembers.removeAll(oldMembers);
    if(newMembers.size() > 0){
        for(Address newAdd : newMembers){
            log.info("Node joined: " + newAdd.toString());
        }
    }
}

MergeEvent の実行中に同様の論理を使用して、クラスターの新しいメンバーセットを判断することもできます。

デフォルトでは、Red Hat JBoss Data Grid の通知はイベントが生成された同じスレッドで送信されます。そのため、スレッドの進行を妨げないようにリスナーを書く必要があります。

この代わりに、別のスレッドで通知を送信し、元のスレッドの操作を妨げないようにするために、リスナーを非同期としてアノテーション付けすることもできます。

以下を使用してリスナーにアノテーションを付けます。

@Listener (sync = false)
public class MyAsyncListener { .... }

XML 設定ファイルで asyncListenerExecutor 要素を使用し、非同期通知の送信に使用されるスレッドプールを調整します。

キャッシュエントリーの作成後、プログラムを使用してキャッシュエントリーを変更できます。

キャッシュエントリーが変更されたリスナーイベントでは、getValue() メソッドの動作は、実際の操作の実行前または実行後にコールバックがトリガーされたかどうかに特定されます。たとえば、event.isPre() が true の場合、event.getValue() は変更の前に古い値を返します。event.isPre() が false の場合、event.getValue() は新しい値を返します。イベントによって新しいエントリーが作成および挿入される場合、古い値は null になります。isPre() の詳細は、Red Hat JBoss Data Grid の『API Documentation』に記載されている org.infinispan.notifications.cachelistener.event パッケージのリストを参照してください。

Listenable および FilteringListenable インターフェース (Cache オブジェクトによって実装される) によって公開されるメソッドを使用する場合のみ、プログラムを使用してリスナーを設定できます。

以下の例は、キャッシュエントリーが変更されるたびに情報を出力する Red Hat JBoss Data Grid のリスナーを定義します。

@Listener
public class PrintWhenModified {

  @CacheEntryModified
  public void print(CacheEntryModifiedEvent event) {
    System.out.println("Cache entry modified. Details = " + event);
  }

}

クラスター化リスナーは、リスナーを分散キャッシュ設定で使用できるようにします。分散キャッシュ環境では、イベントが発生したノードのローカルイベントのみが登録されたローカルリスナーに通知されます。クラスター化リスナーでは、イベントが発生したノードに関係なくクラスターで発生する書き込み通知を 1 つのリスナーが受信できるようにし、この問題に対応します。そのため、クラスター化リスナーのパフォーマンスは、イベントが発生するノードのイベント通知のみを提供する非クラスター化リスナーよりも遅くなります。

クラスター化リスナーを使用する場合、特定のキャッシュでエントリーの追加、更新、期限切れ、または削除が行われるとクライアントアプリケーションに通知が送られます。イベントはクラスター全体であるため、アプリケーションが存在するノードや接続するノードに関係なく、クライアントアプリケーションはイベントにアクセスできます。

イベントは、常にリスナーが登録されたノード上で発生し、キャッシュの更新が発生した場所は関係しません。

以下のユースケースではリスナーは受け取ったイベントを保存します。

@Listener(clustered = true)
  protected static class ClusterListener {
     List<CacheEntryEvent> events = Collections.synchronizedList(new ArrayList<CacheEntryEvent>());

     @CacheEntryCreated
     @CacheEntryModified
     @CacheEntryExpired
     @CacheEntryRemoved
     public void onCacheEvent(CacheEntryEvent event) {
        log.debugf("Adding new cluster event %s", event);
        events.add(event);
     }
  }

  public void addClusterListener(Cache<?, ?> cache) {
     ClusterListener clusterListener = new ClusterListener();
     cache.addListener(clusterListener);
  }

クラスター化リスナーを使用すると、非クラスターリスナーでは適用されない以下の制限が適用されます。

addListener メソッドを使用すると、クラスター化されたリスナーを既存の @CacheListener API の上に追加できます。

cache.addListener(Object listener, Filter filter, Converter converter);

public @interface Listener {
  boolean clustered() default false;
  boolean includeCurrentState() default false;
  boolean sync() default true;
}

interface CacheEventFilter<K,V> {
  public boolean accept(K key, V oldValue, Metadata oldMetadata, V newValue, Metadata newMetadata, EventType eventType);
}

interface CacheEventConverter<K,V,C> {
  public C convert(K key, V oldValue, Metadata oldMetadata, V newValue, Metadata newMetadata, EventType eventType);
}

クラスター化リスナーを使用する場合、キャッシュが更新される順序は通知を受け取る順序に反映されます。

クラスター化リスナーは、イベントが 1 度だけ送信されることを保証しません。同じイベントが複数回送信されないようにするため、リスナー実装はべき等である必要があります。安定したクラスターや、includeCurrentState の結果として合成イベントが生成されるタイムスパン外部では、単一性が受け入れられることをインプリメンターは想定することができます。

次のユースケースは、ニューヨーク州のニューヨーク市宛の注文がいつ生成されるか知りたいリスナーを表しています。リスナーには、ニューヨークから出入りする注文をフィルターする Filter が必要です。さらに、注文全体は必要ではなく、配達される日付のみが必要であるため、Converter も必要になります。

class CityStateFilter implements CacheEventFilter<String, Order> {
    private String state;
    private String city;

    public boolean accept(String orderId, Order oldOrder,
                          Metadata oldMetadata, Order newOrder,
                          Metadata newMetadata, EventType eventType) {
        switch (eventType.getType()) {
            // Only send update if the order is going to our city
            case CACHE_ENTRY_CREATED:
                return city.equals(newOrder.getCity()) &&
                        state.equals(newOrder.getState());
            // Only send update if our order has changed from our city to elsewhere or if is now going to our city
            case CACHE_ENTRY_MODIFIED:
                if (city.equals(oldOrder.getCity()) &&
                        state.equals(oldOrder.getState())) {
                    // If old city matches then we have to compare if new order is no longer going to our city
                    return !city.equals(newOrder.getCity()) ||
                            !state.equals(newOrder.getState());
                } else {
                    // If the old city doesn't match ours then only send update if new update does match ours
                    return city.equals(newOrder.getCity()) &&
                            state.equals(newOrder.getState());
                }
                // On remove we have to send update if our order was originally going to city
            case CACHE_ENTRY_REMOVED:
                return city.equals(oldOrder.getCity()) &&
                        state.equals(oldOrder.getState());
        }
        return false;
    }
}

class OrderDateConverter implements CacheEventConverter<String, Order, Date> {
    private String state;
    private String city;

    public Date convert(String orderId, Order oldValue,
                        Metadata oldMetadata, Order newValue,
                        Metadata newMetadata, EventType eventType) {
        // If remove we do not care about date - this tells listener to remove its data
        if (eventType.isRemove()) {
            return null;
        } else if (eventType.isModified()) {
            if (state.equals(newValue.getState()) &&
                    city.equals(newValue.getCity())) {
                // If it is a modification meaning the destination has changed to ours then we allow it
                return newValue.getDate();
            } else {
                // If destination is no longer our city it means it was changed from us so send null
                return null;
            }
        } else {
            // This was a create so we always send date
            return newValue.getDate();
        }
    }
}

「クラスター化リスナーの例」では、結果のフィルターおよび変換を一度に実行するために、最適化された CacheEventFilterConverter を使用できます。

CacheEventFilterConverter は、イベントのフィルターと変換を一度に実行できるようにする最適化です。これは、イベントフィルターとコンバーターが同じオブジェクトとして最も効率的に使用され、同じメソッドでフィルターと変換が行われる場合に使用できます。戻り値が null の場合は値がフィルターを通過しなかったことを意味するため、変換が null 値を返さない場合のみ使用できます。null 値を変換するには、CacheEventFilter および CacheEventConverter インターフェースを独立して使用します。

以下は、CacheEventFilterConverter を使用したニューヨーク宛の注文のユースケース例になります。

class OrderDateFilterConverter extends AbstractCacheEventFilterConverter<String, Order, Date> {
    private final String state;
    private final String city;

    public Date filterAndConvert(String orderId, Order oldValue,
                                 Metadata oldMetadata, Order newValue,
                                 Metadata newMetadata, EventType eventType) {
        // Remove if the date is not required - this tells listener to remove its data
        if (eventType.isRemove()) {
            return null;
        } else if (eventType.isModified()) {
            if (state.equals(newValue.getState()) &&
                city.equals(newValue.getCity())) {
                // If it is a modification meaning the destination has changed to ours then we allow it
                return newValue.getDate();
            } else {
                // If destination is no longer our city it means it was changed from us so send null
                return null;
            }
        } else {
            // This was a create so we always send date
            return newValue.getDate();
        }
    }
}

リスナーの登録時、フィルターとコンバーターの両方の引数として FilterConverter を提供します。

OrderDateFilterConverter filterConverter = new OrderDateFilterConverter("NY", "New York");
cache.addListener(listener, filterConveter, filterConverter);

イベントリスナーは、Red Hat JBoss Data Grid Hot Rod サーバーが CacheEntryCreated、CacheEntryModified、CacheEntryExpired、CacheEntryRemoved などのイベントのリモートクライアントを通知できるようにします。接続されたクライアントが殺到しないようにするため、クライアントはこれらのイベントをリッスンするかどうかを選択することができます。クライアントがサーバーへの永続接続を維持することが前提となります。

リモートイベントのクライアントリスナーは、ライブラリーモードのクラスター化リスナーと同様に追加できます。以下は、受け取った各イベントを出力するリモートクライアントリスナーの例になります。

import org.infinispan.client.hotrod.annotation.*;
import org.infinispan.client.hotrod.event.*;

@ClientListener
public class EventLogListener {

   @ClientCacheEntryCreated
   public void handleCreatedEvent(ClientCacheEntryCreatedEvent e) {
      System.out.println(e);
   }

   @ClientCacheEntryModified
   public void handleModifiedEvent(ClientCacheEntryModifiedEvent e) {
      System.out.println(e);
   }

   @ClientCacheEntryExpired
   public void handleExpiredEvent(ClientCacheEntryExpiredEvent e) {
      System.out.println(e);
   }

   @ClientCacheEntryRemoved
   public void handleRemovedEvent(ClientCacheEntryRemovedEvent e) {
      System.out.println(e);
   }

}

RemoteCache<Integer, String> cache = rcm.getCache();
cache.addClientListener(new EventLogListener());

EventLogListener listener = ...
cache.removeClientListener(listener);

以下は、Hot Rod 経由でリモートキャッシュと対話するようにリモートクライアントリスナーを設定するために必要な手順になります。

ClientCacheEntryCreatedEvent(key=1,dataVersion=1)
ClientCacheEntryModifiedEvent(key=1,dataVersion=2)
ClientCacheEntryRemovedEvent(key=1)

出力は、デフォルトでイベントにはキーと現在の値に関連するデータバージョンが含まれることを示しています。実際の値はパフォーマンス上の理由でクライアントへ返送されません。リモートイベントの受け取りはパフォーマンスに影響します。キャッシュサイズが大きくなるとより多くの操作が実行されるため、パフォーマンスへの影響も大きくなります。Hot Rod クライアントにイベントが大量に送られないようにするには、サーバー側でリモートイベントをフィルターするか、イベントコンテンツをカスタマイズします。

イベントをフィルターまたはカスタマイズする際、KeyValueFilter および Converter インスタンスはマーシャル可能でなければなりません。クライアントリスナーがクラスターにインストールされると、イベントの送信元でフィルターと変換が行われるようにするため、フィルターインスタンスとコンバーターインスタンスはクラスターの他のノードに送信されます。これにより効率がよくなります。これらのクラスをマーシャル可能にするには、これらのクラスが Serializable を拡張するようにするか、カスタム Externalizer を提供および登録します。

Marshaller インスタンスをサーバー側にデプロイするには、フィルターおよびカスタマイズされたイベントに使用されるメソッドに似たメソッドを使用します。

マーシャラーは、個別の jar または CacheEventConverter や CacheEventFilter インスタンスと同じ jar にデプロイすることができます。

クライアントがリモートリスナーを追加するとき、クラスターの単一のノードにインストールされます。このノードは、クラスター全体で発生する影響受ける操作すべてに対してイベントをクライアントに返送します。

クラスター化された環境では、リスナーが含まれるノードがダウンした場合、Hot Rod クライアント実装が透過的にクライアントリスナーの登録を別のノードにフェイルオーバーします。これにより、イベントの消費にギャップが生じることがありますが、以下の方法の 1 つを使用すると解決できます。

フェイルオーバーの間にクライアントが一部のイベントを見逃すことがあります。イベントを見逃さないようにするには、includeCurrentState パラメーターを true に設定します。これを有効にすると、クライアントはデータを消去でき、 CacheEntryCreatedEvent インスタンスすべてを受信し、すべてのキーでこれらのイベントをキャッシュできます。この代わりに、コールバックハンドラーを追加して、Hot Rod クライアントがフェイルオーバーイベントを認識できるようにすることができます。このコールバックメソッドは、クライアントリスナーに影響するクラスタートポロジーの変更に対応する効率的な方法で、クライアントリスナーはフェイルオーバーでどのように動作するかを判断することができます。ニアキャッシュはこの方法を利用し、ClientCacheFailoverEvent の受信後にニアキャッシュは消去されます。

import org.infinispan.client.hotrod.annotation.*;
import org.infinispan.client.hotrod.event.*;

@ClientListener
public class EventLogListener {
// ...

    @ClientCacheFailover
    public void handleFailover(ClientCacheFailoverEvent e) {
      // Deal with client failover, e.g. clear a near cache.
    }
}