热 Rod Java 客户端指南

如果您不想使用公共 Red Hat Enterprise Maven 存储库，将 Data Grid Maven 存储库下载并安装到本地文件系统、Apache HTTP 服务器或 Maven 存储库管理器。

在您的 Maven 构建环境中包括红帽 GA 存储库，以获取 Data Grid 工件和依赖项。

Maven 使用名为 Project Object Model (POM)文件的配置文件来定义项目并管理构建。POM 文件采用 XML 格式，描述生成的项目打包和输出的模块和组件依赖项、构建顺序和目标。

以下示例显示了 Data Grid 版本和 BOM：

<properties>
  <version.infinispan>11.0.9.Final-redhat-00001</version.infinispan>
</properties>

<dependencyManagement>
  <dependencies>
    <dependency>
      <groupId>org.infinispan</groupId>
      <artifactId>infinispan-bom</artifactId>
      <version>${version.infinispan}</version>
      <type>pom</type>
      <scope>import</scope>
    </dependency>
  </dependencies>
</dependencyManagement>

使用拓扑感知智能的客户端对所有请求使用请求平衡。默认平衡策略是 round-robin，因此拓扑感知客户端始终以轮循顺序向服务器发送请求。

例如，s1、 s2、 s3 是 Data Grid 集群中的服务器。客户端执行请求平衡，如下所示：

CacheContainer cacheContainer = new RemoteCacheManager();
Cache<String, String> cache = cacheContainer.getCache();

//client sends put request to s1
cache.put("key1", "aValue");
//client sends put request to s2
cache.put("key2", "aValue");
//client sends get request to s3
String value = cache.get("key1");
//client dispatches to s1 again
cache.remove("key2");
//and so on...

使用分布感知智能的客户端仅对失败的请求使用请求平衡。当请求失败时，分发感知客户端会在下一个可用服务器上重试请求。

infinispan.client.hotrod.request_balancing_strategy

当 Data Grid 集群拓扑更改时，热 Rod 客户端可以自动进行故障转移。例如，具有拓扑感知的 Hot Rod 客户端可以检测一个或多个数据网格服务器何时失败。

除了集群数据网格服务器之间故障转移外，Hot Rod 客户端还可在 Data Grid 集群之间进行故障转移。

例如，您有一个 Data Grid 集群在 New York (NYC)中运行，另一个集群在伦敦(LON)中运行。向 NYC 发送请求的客户端检测到没有可用的节点，以便它们在 LON 中切换到集群。然后，客户端会维护到 LON 的连接，直到您手动切换集群或故障转移再次发生。

虽然 Hot Rod 客户端可以故障转移，但您应该使用事务缓存来确保操作不会部分完成，并在不同的节点上保留冲突的条目。

Data Grid 支持带有 Hot Rod 连接器的以下 SASL 验证机制：

要使用 GSSAPI 机制，您必须创建一个 LoginContext，以便您的 Hot Rod 客户端可以获取 Ticket Granting Ticket Granting Ticket Granting Ticket Granting Ticket Granting Ticket (TGT)。

在 Data Grid 集群间手动切换 Hot Rod Java 客户端连接。

您可以将 ProtoStream SerializationContextInitializer 接口的实现添加到 Hot Rod 客户端配置，以便 Data Grid marshalls 自定义 Java 对象。

infinispan.client.hotrod.context-initializers=org.infinispan.example.LibraryInitializerImpl,org.infinispan.example.AnotherExampleSciImpl

ConfigurationBuilder builder = new ConfigurationBuilder();
builder
      .addServer()
      .host("127.0.0.1")
      .port(11222)
      .addContextInitializers(new LibraryInitializerImpl(), new AnotherExampleSciImpl());
RemoteCacheManager rcm = new RemoteCacheManager(builder.build());

将 Hot Rod 客户端配置为使用自定义 marshallers。

Data Grid RemoteCache API 不支持 Cache API 中的所有方法，并在调用不支持的方法时抛出 UnsupportedOperationException。

大多数方法不适用于远程缓存（如侦听器管理操作），或者对应于本地缓存不支持的方法（例如，containsValue）。

RemoteCache API 不支持从 ConcurrentMap 继承的某些原子操作，例如：

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");

默认情况下生成的事件仅包含足够的信息，以便使事件相关，但可以避免产生太多的信息，以降低发送它们的成本。（可选）事件中提供的信息可以自定义，使其包含更多信息，如值，或者包含较少的信息。此自定义通过 CacheEventConverter Factory 生成的 CacheEventConverterFactory 实例进行：

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，它允许过滤和自定义在一个步骤中进行。为方便起见，建议直接扩展 org.infinispan.notifications.cachelistener.filter.AbstractCacheEventFilterConverter，而不是直接实施 org.infinispan.notifications.cachelistener.filter.CacheEventFilterConverter。例如：

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);
   }
}

与过滤器和转换器类似，要能够使用组合的 filter/converter 工厂注册监听程序，工厂必须通过 @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);

热 Rod 服务器可以以不同的格式存储数据，但是尽管如此，Java Hot Rod 客户端用户仍然可以开发在键入的对象上运行的 CacheEventConverter 或 CacheEventFilter 实例。默认情况下，过滤器和转换器将数据用作 POJO (application/x-java-object)，但可以通过覆盖 filter/converter 中的方法 格式（） 来覆盖所需的格式。如果格式返回 null，则过滤器/转换器将接收存储的数据。

热 Rod Java 客户端可以配置为使用不同的 org.infinispan.commons.marshall.Marshaller 实例。如果这样做和部署 CacheEventConverter 或 CacheEventFilter 实例，则可以使用 Java 对象而不是 marshaller 显示过滤器/转换，服务器需要能够在对象和 marshaller 生成的二进制格式之间进行转换。

要部署 Marshaller 实例服务器端，请按照类似的方法部署 CacheEventConverter 或 CacheEventFilter 实例：

请注意，Marshaller 可以部署到单独的 jar 中，或者在与 CacheEventConverter 和/或 CacheEventFilter 实例相同的 jar 中进行部署。

客户端侦听器注解具有可选的 includeCurrentState 属性，用于指定在添加监听程序时是否将状态发送到客户端，或者是监听器故障转移时。

默认情况下，includeCurrentState 为 false，但如果设置为 true，并且客户端监听程序添加到已包含数据的缓存中，服务器会迭代缓存内容，并将每个条目的事件作为 ClientCacheEntryCreated （如果配置）发送一个 ClientCacheEntryCreated （如果配置了自定义事件）。这允许客户端基于现有内容构建一些本地数据结构。迭代内容后，事件会正常接收，因为接收缓存更新。如果缓存被集群，则整个集群范围的内容都会迭代。

当 Hot Rod 客户端注册客户端监听程序时，它会在集群的单个节点中执行此操作。如果该节点失败，Java Hot Rod 客户端会检测到透明且在节点中注册的所有监听器失败。

在这种故障切换过程中，客户端可能会错过一些事件。为了避免缺少这些事件，client 侦听器注解包含一个名为 includeCurrentState 的可选参数，如果设为 true，则缓存内容可以迭代，并且生成 ClientCacheEntryCreated 事件（如果配置了自定义事件）。默认情况下，includeCurrentState 设置为 false。

使用回调来处理故障转移事件：

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

当客户端缓存了一些数据的用例中，这非常有用，因此，考虑到一些事件可能会丢失，它决定在收到事件失败时清除任何本地缓存的数据，了解事件故障转移后，它将收到整个缓存的内容的事件。

服务器中的缓存还必须是事务处理，客户端才能参与 JTA {tx}s。

需要以下服务器配置，否则只进行事务回滚：

例如：

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

热 Rod 事务有自己的恢复机制。

在创建 {rcm} 时，您可以设置 {rc} 使用的默认 {tm} 和 {tx-mode}。

{rcm} 可让您只为事务缓存创建一个配置，如下例所示：

org.infinispan.client.hotrod.configuration.ConfigurationBuilder cb = new org.infinispan.client.hotrod.configuration.ConfigurationBuilder();
//other client configuration parameters
cb.transaction().transactionManagerLookup(GenericTransactionManagerLookup.getInstance());
cb.transaction().transactionMode(TransactionMode.NON_XA);
cb.transaction().timeout(1, TimeUnit.MINUTES)
RemoteCacheManager rmc = new RemoteCacheManager(cb.build());

前面的配置应用到远程缓存的所有实例。如果需要将不同的配置应用到远程缓存实例，您可以覆盖 {rc} 配置。请参阅 覆盖 RemoteCacheManager 配置。

有关配置参数的文档，请参阅 {cb} Javadoc。

您还可以使用属性文件配置 Java Hot Rod 客户端，如下例所示：

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

{tx-mode} 控制 {rc} 与 {tm} 交互的方式。

在 Data Grid 配置和客户端设置中，事务模式都相同。将以下模式与客户端搭配使用，请参阅服务器的 Data Grid 配置模式：

因为 {rcm} 不支持每个缓存实例的不同配置。但是，{rcm} 包含返回 {rc} 实例的 getCache (String) 方法，可让您覆盖一些配置参数，如下所示：

事务使用键的初始值来检测冲突。

例如，当事务开始时，"k" 的值为 "v"。在准备阶段，事务从服务器获取"k"以读取值。如果值已更改，事务会回滚以避免冲突。

forceReturnValue 参数控制对 {rc} 的写入操作，并帮助避免冲突。它有以下值：

以下事务提供了一个示例，其中 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.get （） 操作，以便在执行 cache.put （） 操作前读取 "k" 的值：

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 参数不会生效。一个事务提交，另一个回滚。但是 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.transaction().transactionManagerLookup(RemoteTransactionManagerLookup.getInstance());
cb.transaction().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();

以下示例演示了如何使用 getCache 方法覆盖 TransactionManager ：

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

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

//Define a custom TransactionManager.
TransactionManager myCustomTM = ...

//Override the TransactionManager for the my-cache instance. Use the default configuration if null is returned.
RemoteCache<String, String> cache = rcm.getCache("my-cache", null, myCustomTM);

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

以下示例演示了如何使用 getCache 方法覆盖 TransactionMode ：

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

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

//Override the transaction mode for the my-cache instance.
RemoteCache<String, String> cache = rcm.getCache("my-cache", TransactionMode.NON_DURABLE_XA, null);

//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();