Hot Rod Java クライアントガイド

Hot Rod クライアント設定で Data Grid クラスターの場所を指定します。

ConfigurationBuilder clientBuilder = new ConfigurationBuilder();
clientBuilder.addCluster("siteA")
               .addClusterNode("hostA1", 11222)
               .addClusterNode("hostA2", 11222)
             .addCluster("siteB")
               .addClusterNodes("hostB1:11222; hostB2:11222");
RemoteCacheManager remoteCacheManager = new RemoteCacheManager(clientBuilder.build());

ConfigurationBuilder clientBuilder = new ConfigurationBuilder();
clientBuilder.addServers("hostA1:11222; hostA2:11222")
             .addCluster("siteB")
               .addClusterNodes("hostB1:11222; hostB2:11223");
RemoteCacheManager remoteCacheManager = new RemoteCacheManager(clientBuilder.build());

Data Grid クラスター間で Hot Rod Java クライアント接続を手動で切り替えます。

Hot Rod Java クライアントは、Data Grid サーバーへの永続的な接続のプールを保持して、要求ごとに TCP 接続を作成するのではなく、TCP 接続を再利用します。

ConfigurationBuilder clientBuilder = new ConfigurationBuilder();
clientBuilder.addServer()
               .host("127.0.0.1")
               .port(11222)
             .connectionPool()
               .maxActive(10)
               .exhaustedAction(ExhaustedAction.valueOf("WAIT"))
               .maxWait(1)
               .minIdle(20)
               .minEvictableIdleTime(300000)
               .maxPendingRequests(20);
RemoteCacheManager remoteCacheManager = new RemoteCacheManager(clientBuilder.build());

infinispan.client.hotrod.server_list = 127.0.0.1:11222
infinispan.client.hotrod.connection_pool.max_active = 10
infinispan.client.hotrod.connection_pool.exhausted_action = WAIT
infinispan.client.hotrod.connection_pool.max_wait = 1
infinispan.client.hotrod.connection_pool.min_idle = 20
infinispan.client.hotrod.connection_pool.min_evictable_idle_time = 300000
infinispan.client.hotrod.connection_pool.max_pending_requests = 20

これらの設定オプションにより、プールをきめ細かく制御できます。アプリケーションのニーズに応じて、プールをチューニングすることを推奨します。そうしないと、新しい接続を開くときに待ち時間が増加したり、利用可能な接続が少なくなってスループットが低下したりする可能性があります。

ロードが急増すると、大きなプールは新しい接続で「群集効果」が発生する可能性があります。ただし、リクエストに対応する接続が増えると、スループットが向上するはずです。小さいプールはこの影響を回避するため、消費されるリソースが少なくなります。ただし、処理できるリクエスト数が制限されるため、上限に達するとパフォーマンスが低下し始めます。

GSSAPI メカニズムを使用するには、LoginContextを作成して、Hot Rod クライアント Ticket Granting Ticket (TGT) を取得できるようにする必要があります。

Data Grid Server は、Hot Rod および Memcached バイナリープロトコルエンドポイントを使用して、次の SASL 認証メカニズムをサポートします。

ニアキャッシュで Hot Rod Java クライアントを設定し、最近使用されたデータをクライアント JVM に保存します。

import org.infinispan.client.hotrod.configuration.ConfigurationBuilder;
import org.infinispan.client.hotrod.configuration.NearCacheMode;
import org.infinispan.client.hotrod.configuration.ExhaustedAction;

ConfigurationBuilder builder = new ConfigurationBuilder();
builder.addServer()
    .host("127.0.0.1")
    .port(ConfigurationProperties.DEFAULT_HOTROD_PORT)
  .security().authentication()
    .username("username")
    .password("password")
    .realm("default")
    .saslMechanism("SCRAM-SHA-512")
  // Configure the connection pool for bloom filters.
  .connectionPool()
    .maxActive(1)
    .exhaustedAction(ExhaustedAction.WAIT);
// Configure near caching for specific caches
builder.remoteCache("books")
    .nearCacheMode(NearCacheMode.INVALIDATED)
    .nearCacheMaxEntries(100)
    .nearCacheUseBloomFilter(false);
builder.remoteCache("authors")
    .nearCacheMode(NearCacheMode.INVALIDATED)
    .nearCacheMaxEntries(200)
    .nearCacheUseBloomFilter(true);

Data Grid の RemoteCache API は、Cache API で利用可能なすべてのメソッドをサポートしておらず、サポート対象外のメソッドが呼び出されると UnsupportedOperationException を出力します。

これらのメソッドのほとんどはリモートキャッシュ (リスナー管理操作など) では意味を持ちません。または、ローカルキャッシュでサポートされていないメソッドにも対応しません (例: containsValue)。

ConcurrentMap から継承された特定のアトミック操作も、RemoteCache API ではサポートされていません。以下に例を示します。

boolean remove(Object key, Object value);
boolean replace(Object key, Object value);
boolean replace(Object key, Object oldValue, Object value);

ただし、RemoteCache は、値オブジェクト全体ではなく、ネットワーク経由でバージョン識別子を送信する、これらのアトミック操作の代替バージョンメソッドを提供します。

カスタムフィルターを Data Grid サーバーインスタンスにデプロイします。

クライアントイベントリスナーが必要ない場合は、以下を削除できます。

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

クライアントにイベントが殺到するのを防ぐために、ユーザーはフィルタリング機能を提供して、特定のクライアントリスナーに対し、サーバーによって発生するイベントの数を制限できます。フィルターを有効にするには、フィルターインスタンスを生成するキャッシュイベントフィルターファクトリーを作成する必要があります。

import org.infinispan.notifications.cachelistener.filter.CacheEventFilterFactory;
import org.infinispan.filter.NamedFactory;

@NamedFactory(name = "static-filter")
public static class StaticCacheEventFilterFactory implements CacheEventFilterFactory {

   @Override
   public StaticCacheEventFilter getFilter(Object[] params) {
      return new StaticCacheEventFilter();
   }
}


// Serializable, Externalizable or marshallable with Infinispan Externalizers
// needed when running in a cluster
class StaticCacheEventFilter implements CacheEventFilter<Integer, String>, Serializable {
   @Override
   public boolean accept(Integer key, String oldValue, Metadata oldMetadata,
         String newValue, Metadata newMetadata, EventType eventType) {
      if (key.equals(1)) // static key
         return true;

      return false;
   }
}

上記で定義されたキャッシュイベントフィルターファクトリーインスタンスは、キーが 1 であるものを除くすべてのエントリーを静的にフィルターするインスタンスを作成します。

このキャッシュイベントフィルターファクトリーにリスナーを登録できるようにするには、ファクトリーに一意の名前を付ける必要があり、Hot Rod サーバーに名前とキャッシュイベントフィルターファクトリーインスタンスを接続する必要があります。

リスナーの登録時に提供されたパラメーターに基づいてフィルタリングする動的フィルターインスタンスを登録することもできます。フィルターは、フィルターファクトリーが受け取ったパラメーターを使用してこのオプションを有効にします。以下に例を示します。

import org.infinispan.notifications.cachelistener.filter.CacheEventFilterFactory;
import org.infinispan.notifications.cachelistener.filter.CacheEventFilter;

class DynamicCacheEventFilterFactory implements CacheEventFilterFactory {
   @Override
   public CacheEventFilter<Integer, String> getFilter(Object[] params) {
      return new DynamicCacheEventFilter(params);
   }
}

// Serializable, Externalizable or marshallable with Infinispan Externalizers
// needed when running in a cluster
class DynamicCacheEventFilter implements CacheEventFilter<Integer, String>, Serializable {
   final Object[] params;

   DynamicCacheEventFilter(Object[] params) {
      this.params = params;
   }

   @Override
   public boolean accept(Integer key, String oldValue, Metadata oldMetadata,
         String newValue, Metadata newMetadata, EventType eventType) {
      if (key.equals(params[0])) // dynamic key
         return true;

      return false;
   }
}

リスナーの登録時に、フィルタリングに必要な動的パラメーターが提供されます。

RemoteCache<?, ?> cache = ...
cache.addClientListener(new EventPrintListener(), new Object[]{1}, null);

サーバーからイベント通知を取得しなくても、リモート API メソッドを呼び出してオペレーションを実行する際に、SKIP_LISTENER_NOTIFICATION フラグを含めます。たとえば、値を作成または変更する際のリスナーの通知を防ぐには、フラグを以下のように設定します。

remoteCache.withFlags(Flag.SKIP_LISTENER_NOTIFICATION).put(1, "one");

デフォルトで生成されるイベントには、イベントを関連させるのに十分な情報が含まれていますが、送信コストを削減するために情報を詰め込みすぎないようにしています。必要に応じて、イベントに含まれる情報は、値などの詳細情報を含むか、より少ない情報を含めるためにカスタム化できます。このカスタマイズは、CacheEventConverterFactory によって生成された CacheEventConverter インスタンスを使用して行われます。

import org.infinispan.notifications.cachelistener.filter.CacheEventConverterFactory;
import org.infinispan.notifications.cachelistener.filter.CacheEventConverter;
import org.infinispan.filter.NamedFactory;

@NamedFactory(name = "static-converter")
class StaticConverterFactory implements CacheEventConverterFactory {
   final CacheEventConverter<Integer, String, CustomEvent> staticConverter = new StaticCacheEventConverter();
   public CacheEventConverter<Integer, String, CustomEvent> getConverter(final Object[] params) {
      return staticConverter;
   }
}

// Serializable, Externalizable or marshallable with Infinispan Externalizers
// needed when running in a cluster
class StaticCacheEventConverter implements CacheEventConverter<Integer, String, CustomEvent>, Serializable {
   public CustomEvent convert(Integer key, String oldValue, Metadata oldMetadata, String newValue, Metadata newMetadata, EventType eventType) {
      return new CustomEvent(key, newValue);
   }
}

// Needs to be Serializable, Externalizable or marshallable with Infinispan Externalizers
// regardless of cluster or local caches
static class CustomEvent implements Serializable {
   final Integer key;
   final String value;
   CustomEvent(Integer key, String value) {
      this.key = key;
      this.value = value;
   }
}

上記の例では、コンバーターは、イベント内の値とキーを含む新しいカスタムイベントを生成します。これにより、デフォルトのイベントと比べて大量のイベントペイロードが発生しますが、フィルターと組み合わせると、ネットワークの帯域幅コストを減らすことができます。

カスタムイベントの処理には、以前に実証されたものに対して、若干異なるクライアントリスナーの実装が必要になります。より正確な状態に戻すには、ClientCacheEntryCustomEvent インスタンスを処理する必要があります。

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

@ClientListener
public class CustomEventPrintListener {

   @ClientCacheEntryCreated
   @ClientCacheEntryModified
   @ClientCacheEntryRemoved
   public void handleCustomEvent(ClientCacheEntryCustomEvent<CustomEvent> e) {
      System.out.println(e);
   }

}

コールバックで受け取った ClientCacheEntryCustomEvent は、getEventData メソッドを介してカスタムイベントを公開し、getType メソッドは、生成されたイベントがキャッシュエントリーの作成、変更、または削除の結果であったかどうかに関する情報を提供します。

フィルタリングと同様に、リスナーをこのコンバーターファクトリーに登録できるようにするには、ファクトリーに一意の名前を付ける必要があり、Hot Rod サーバーに名前とキャッシュイベントコンバーターファクトリーインスタンスを接続する必要があります。

リスナーの登録時に提供されたパラメーターを基に変換する動的コンバーターインスタンスもあります。コンバーターは、コンバーターファクトリーによって受け取ったパラメーターを使用してこのオプションを有効にします。以下に例を示します。

import org.infinispan.notifications.cachelistener.filter.CacheEventConverterFactory;
import org.infinispan.notifications.cachelistener.filter.CacheEventConverter;

@NamedFactory(name = "dynamic-converter")
class DynamicCacheEventConverterFactory implements CacheEventConverterFactory {
   public CacheEventConverter<Integer, String, CustomEvent> getConverter(final Object[] params) {
      return new DynamicCacheEventConverter(params);
   }
}

// Serializable, Externalizable or marshallable with Infinispan Externalizers needed when running in a cluster
class DynamicCacheEventConverter implements CacheEventConverter<Integer, String, CustomEvent>, Serializable {
   final Object[] params;

   DynamicCacheEventConverter(Object[] params) {
      this.params = params;
   }

   public CustomEvent convert(Integer key, String oldValue, Metadata oldMetadata,
         String newValue, Metadata newMetadata, EventType eventType) {
      // If the key matches a key given via parameter, only send the key information
      if (params[0].equals(key))
         return new CustomEvent(key, null);

      return new CustomEvent(key, newValue);
   }
}

変換を行うために必要な動的パラメーターは、リスナーが登録されたときに提供されます。

RemoteCache<?, ?> cache = ...
cache.addClientListener(new EventPrintListener(), null, new Object[]{1});

イベントのフィルタリングとカスタマイズの両方を実行する場合、org.infinispan.notifications.cachelistener.filter.CacheEventFilterConverter を実装する方が簡単です。これにより、1 つのステップでフィルターとカスタマイズの両方を実行できます。便宜上、org.infinispan.notifications.cachelistener.filter.CacheEventFilterConverter を直接実装するのではなく、org.infinispan.notifications.cachelistener.filter.AbstractCacheEventFilterConverter を拡張することが推奨されます。以下に例を示します。

import org.infinispan.notifications.cachelistener.filter.CacheEventConverterFactory;
import org.infinispan.notifications.cachelistener.filter.CacheEventConverter;

@NamedFactory(name = "dynamic-filter-converter")
class DynamicCacheEventFilterConverterFactory implements CacheEventFilterConverterFactory {
   public CacheEventFilterConverter<Integer, String, CustomEvent> getFilterConverter(final Object[] params) {
      return new DynamicCacheEventFilterConverter(params);
   }
}

// Serializable, Externalizable or marshallable with Infinispan Externalizers needed when running in a cluster
//
class DynamicCacheEventFilterConverter extends AbstractCacheEventFilterConverter<Integer, String, CustomEvent>, Serializable {
   final Object[] params;

   DynamicCacheEventFilterConverter(Object[] params) {
      this.params = params;
   }

   public CustomEvent filterAndConvert(Integer key, String oldValue, Metadata oldMetadata,
         String newValue, Metadata newMetadata, EventType eventType) {
      // If the key matches a key given via parameter, only send the key information
      if (params[0].equals(key))
         return new CustomEvent(key, null);

      return new CustomEvent(key, newValue);
   }
}

フィルターとコンバーターと同様に、この組み合わせたフィルター/変換ファクトリーでリスナーを登録するには、ファクトリーに @NamedFactory アノテーション経由で一意の名前が付与される必要があります。また、Hot Rod サーバーは名前とキャッシュイベントコンバーターファクトリーインスタンスと共にプラグインする必要があります。

クライアントの観点からすると、組み合わせたフィルターとコンバータークラスを使用できるようにするには、クライアントリスナーは同じフィルターファクトリーとコンバーターファクトリー名を定義する必要があります。以下に例を示します。

@ClientListener(filterFactoryName = "dynamic-filter-converter", converterFactoryName = "dynamic-filter-converter")
public class CustomEventPrintListener { ... }

上記の例で必要な動的パラメーターは、リスナーが filter パラメーターまたは converter パラメーターのいずれかに登録されている場合に提供されます。フィルターパラメーターが空でない場合は、それらが使用されます。それ以外の場合は、コンバーターパラメーターは以下のようになります。

RemoteCache<?, ?> cache = ...
cache.addClientListener(new CustomEventPrintListener(), new Object[]{1}, null);

Hot Rod サーバーは、異なる形式でデータを保存できますが、Java Hot Rod クライアントユーザーは、タイプ化されたオブジェクトで機能する CacheEventConverter インスタンスまたは CacheEventFilter インスタンスを開発できます。デフォルトでは、フィルターとコンバーターは、データを POJO (application/x-java-object) として使用しますが、filter/converter からメソッド format() を上書きすることで、希望の形式を上書きすることができます。形式が null を返す場合、フィルター/変換は保存時にデータを受け取ります。

Hot Rod Java クライアントは、異なる org.infinispan.commons.marshall.Marshaller インスタンスを使用するように設定できます。これ CacheEventConverter インスタンスや CacheEventFilter インスタンスをデプロイしたり、コンテンツをマーシャリングしたりするのではなく、Java オブジェクトでフィルター/変換できるようにするには、サーバーはオブジェクトとマーシャラーで生成されるバイナリー形式の間で変換できるようにする必要があります。

Marshaller インスタンスのサーバー側のデプロイするには、CacheEventConverter インスタンスまたは CacheEventFilter インスタンスをデプロイするのに使用されるものと同様の方法に従います。

Marshaller は個別の jar または CacheEventConverter インスタンスおよび/または CacheEventFilter インスタンスと同じ jar にデプロイできることに注意してください。

クライアントリスナーアノテーションには、リスナーの追加時またはリスナーのフェイルオーバー時に状態がクライアントに送信されるかどうかを指定する任意の includeCurrentState 属性があります。

デフォルトでは、includeCurrentState は false ですが、true に設定され、クライアントリスナーがデータが含まれるキャッシュに追加されると、サーバーはキャッシュの内容を繰り返し処理し、各エントリーのイベント (設定されている場合はカスタムイベント) を ClientCacheEntryCreated としてクライアントに送信します。これにより、クライアントは既存のコンテンツに基づいて一部のローカルデータ構造をビルドできます。コンテンツが繰り返されると、キャッシュの更新を受け取るため、イベントは通常どおり受信されます。キャッシュがクラスター化されている場合、クラスター全体のコンテンツ全体が繰り返し処理されます。

Hot Rod クライアントがクライアントリスナーを登録すると、クラスターの単一ノードになります。そのノードに障害が発生すると、Java Hot Rod クライアントは透過的に検出し、別のノードに失敗したノードに登録されているすべてのリスナーをフェイルオーバーします。

このフェイルオーバー中に、クライアントはいくつかのイベントを見逃す可能性があります。これらのイベントが不足しないように、クライアントリスナーアノテーションには includeCurrentState という任意のパラメーターが含まれます。これは、フェイルオーバーの実行時に、キャッシュの内容が繰り返し処理され、ClientCacheEntryCreated イベント (設定されている場合はカスタムイベント) が生成されます。デフォルトでは、includeCurrentState は false に設定されています。

コールバックを使用してフェイルオーバーイベントを処理します。

@ClientCacheFailover
public void handleFailover(ClientCacheFailoverEvent e) {
  ...
}

これは、クライアントがいくつかのデータをキャッシュしており、フェイルオーバーの結果、いくつかのイベントが見逃される可能性を考慮して、フェイルオーバーイベントを受信したときにローカルにキャッシュされたデータを消去することを決定し、フェイルオーバーイベントの後にキャッシュ全体のコンテンツに対するイベントを受信することを知っているような場合に非常に便利です。

クライアントが JTA トランザクション に参加するには、サーバーのキャッシュもトランザクションである必要があります。

次のサーバー設定が必要です。それ以外の場合、トランザクションはロールバックのみになります。

以下に例を示します。

<replicated-cache name="hotrodReplTx">
  <locking isolation="REPEATABLE_READ"/>
  <transaction mode="NON_XA" locking="PESSIMISTIC"/>
</replicated-cache>

Hot Rod トランザクションには、独自の復旧メカニズムがあります。

トランザクションの RemoteCache は、キャッシュごとに設定されます。例外は、グローバルなトランザクションの timeout です。これは、単一のトランザクションが複数の RemoteCache と対話できるためです。

以下の例は、キャッシュ my-cache にトランザクションの RemoteCache を設定する方法を示しています。

org.infinispan.client.hotrod.configuration.ConfigurationBuilder cb = new org.infinispan.client.hotrod.configuration.ConfigurationBuilder();
//other client configuration parameters
cb.transactionTimeout(1, TimeUnit.MINUTES);
cb.remoteCache("my-cache")
   .transactionManagerLookup(GenericTransactionManagerLookup.getInstance())
   .transactionMode(TransactionMode.NON_XA);

設定パラメーターの詳細は、ConfigurationBuilder および RemoteCacheConfigurationBuilder Javadoc のドキュメントを参照してください。

以下の例のように、プロパティーファイルを使用して Java Hot Rod クライアントを設定することもできます。

infinispan.client.hotrod.cache.my-cache.transaction.transaction_manager_lookup = org.infinispan.client.hotrod.transaction.lookup.GenericTransactionManagerLookup
infinispan.client.hotrod.cache.my-cache.transaction.transaction_mode = NON_XA
infinispan.client.hotrod.transaction.timeout = 60000

TransactionMode は、RemoteCache が TransactionManager と対話する方法を制御します。

トランザクションモードは、Data Grid 設定とクライアント設定の両方で同じです。クライアントで以下のモードを使用します。サーバーの Data Grid 設定スキーマを参照してください。

トランザクションはキーの初期値を使用して競合を検出します。

たとえば、トランザクションの開始時に "k" の値は "v" となります。準備フェーズでは、トランザクションはサーバーから "k" を取得して値を読み取ります。値が変更された場合、トランザクションは競合を回避するためにロールバックします。

forceReturnValue パラメーターは、RemoteCache への書き込み操作を制御し、競合を回避するのに役立ちます。次の値があります。

以下のトランザクションは、forceReturnValue パラメーターが競合する書き込み操作を防ぐ例を提供します。

RemoteCache<String, String> cache = ...
TransactionManager tm = ...

tm.begin();
cache.put("k", "v1");
tm.commit();

RemoteCache<String, String> cache = ...
TransactionManager tm = ...

tm.begin();
cache.put("k", "v2");
tm.commit();

この例では、TX1 と TX2 が並行して実行されます。"k" の初期値は "v" です。

以下のトランザクションは、cache.put() オペレーションを行う前に "k" の値を読み取る cache.get() オペレーションが含まれます。

RemoteCache<String, String> cache = ...
TransactionManager tm = ...

tm.begin();
cache.get("k");
cache.put("k", "v1");
tm.commit();

RemoteCache<String, String> cache = ...
TransactionManager tm = ...

tm.begin();
cache.get("k");
cache.put("k", "v2");
tm.commit();

上記の例では、TX1 および TX2 の両方が鍵を読み取るため、forceReturnValue パラメーターは有効になりません。1 つのトランザクションがコミットし、もう 1 つのトランザクションがロールバックします。ただし、cache.get() 操作には追加のサーバー要求が必要です。サーバーリクエストが非効率な cache.put() オペレーションの戻り値が必要ない場合。

以下の例は、RemoteCacheManager で設定した TransactionManager および TransactionMode を使用する方法を示しています。

//Configure the transaction manager and transaction mode.
org.infinispan.client.hotrod.configuration.ConfigurationBuilder cb = new org.infinispan.client.hotrod.configuration.ConfigurationBuilder();
cb.remoteCache("my-cache")
    .transactionManagerLookup(RemoteTransactionManagerLookup.getInstance())
    .transactionMode(TransactionMode.NON_XA);

RemoteCacheManager rcm = new RemoteCacheManager(cb.build());

//The my-cache instance uses the RemoteCacheManager configuration.
RemoteCache<String, String> cache = rcm.getCache("my-cache");

//Return the transaction manager that the cache uses.
TransactionManager tm = cache.getTransactionManager();

//Perform a simple transaction.
tm.begin();
cache.put("k1", "v1");
System.out.println("K1 value is " + cache.get("k1"));
tm.commit();