在 Java 应用程序中嵌入了 Data Grid

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

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

配置项目中的项目对象模型(POM)文件，以将 Data Grid 依赖项用于嵌入式缓存、Hot Rod 客户端和其他功能。

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

<properties>
  <version.infinispan>14.0.21.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>

将 Data Grid 添加到项目，以便在您的应用程序中创建嵌入的缓存。

<dependencies>
  <dependency>
    <groupId>org.infinispan</groupId>
    <artifactId>infinispan-core</artifactId>
  </dependency>
</dependencies>

Data Grid 提供了一个 GlobalConfigurationBuilder API，用于控制 Cache Manager 和用于配置缓存的 ConfigurationBuilder API。

// Set up a clustered Cache Manager.
GlobalConfigurationBuilder global = GlobalConfigurationBuilder.defaultClusteredBuilder();
// Initialize the default Cache Manager.
DefaultCacheManager cacheManager = new DefaultCacheManager(global.build());
// Create a distributed cache with synchronous replication.
ConfigurationBuilder builder = new ConfigurationBuilder();
                     builder.clustering().cacheMode(CacheMode.DIST_SYNC);
// Obtain a volatile cache.
Cache<String, String> cache = cacheManager.administration().withFlags(CacheContainerAdmin.AdminFlag.VOLATILE).getOrCreateCache("myCache", builder.build());
// Stop the Cache Manager.
cacheManager.stop();

Cache<String, String> myCache = manager.getCache("myCache");

前面的操作会创建一个名为 myCache 的缓存（如果尚未存在），并返回它。

使用 getCache （） 方法仅在您调用方法的节点上创建缓存。换句话说，它会执行一个本地操作，它必须在集群中的每个节点上调用。通常，在多个节点间部署的应用程序会在初始化过程中获取缓存，以确保缓存都是 对称的，并在每个节点上存在。

Cache<String, String> myCache = manager.administration().createCache("myCache", "myTemplate");

前面的操作还会在随后加入集群的任何节点上自动创建缓存。

默认情况下，您使用 createCache （） 方法创建的缓存是临时的。如果整个集群关闭，则缓存重启后不会自动创建。

Cache<String, String> myCache = manager.administration().withFlags(AdminFlag.PERMANENT).createCache("myCache", "myTemplate");

要使 PERMANENT 标志生效，您必须启用全局状态并设置配置存储提供程序。

有关配置存储供应商的更多信息，请参阅 GlobalStateConfigurationBuilder#configurationStorage （）。

Data Grid 提供了一个 Cache 接口，它公开简单的方法来添加、检索和删除条目，包括 JDK 的 ConcurrentMap 接口公开的原子机制。根据所使用的缓存模式，调用这些方法会触发多个事情，甚至可能包括将条目复制到远程节点或从远程节点查找条目，或可能缓存存储。

对于简单使用，使用 Cache API 不应与使用 JDK Map API 不同，因此从基于映射到 Data Grid 的缓存的简单内存缓存迁移应该很简单。

试图全局执行这些操作会对性能有很大的影响，并成为可扩展性瓶颈。因此，这些方法应该仅用于信息或调试目的。

应注意，在 withFlags （） 方法中使用某些标记可以缓解其中的一些问题，请检查每个方法的文档以了解更多详情。

除了 Lifespan 外，Data Grid 还支持 maxIdle 作为额外的指标，以决定到期。可以使用任何 lifespans 或 maxIdles 的组合。

要达到此目的，putForExternalRead （） 充当 put 调用，只有在缓存中不存在密钥时才运行，并在另一个线程尝试同时存储同一密钥时失败。在这个特殊情况下，缓存数据是优化系统的方法，而不需要缓存失败会影响到持续的事务，因此为什么以不同的方式处理失败。putForExternalRead （） 被视为快速操作，因为无论它是否成功，它都不会等待任何锁定，因此会立即返回到调用者。

要了解如何使用此操作，让我们来看基本的示例。试想一下，个人实例的缓存，每个由 PersonId 的密钥，其数据源自于单独的数据存储中。以下代码显示了在此示例上下文中使用 putForExternalRead 的最常见模式：

// Id of the person to look up, provided by the application
PersonId id = ...;

// Get a reference to the cache where person instances will be stored
Cache<PersonId, Person> cache = ...;

// First, check whether the cache contains the person instance
// associated with with the given id
Person cachedPerson = cache.get(id);

if (cachedPerson == null) {
   // The person is not cached yet, so query the data store with the id
   Person person = dataStore.lookup(id);

   // Cache the person along with the id so that future requests can
   // retrieve it from memory rather than going to the data store
   cache.putForExternalRead(id, person);
} else {
   // The person was found in the cache, so return it to the application
   return cachedPerson;
}

请注意，putForExternalRead 不应用作使用源自应用程序执行的新 Person 实例更新缓存的机制（例如，来自修改人员地址的事务）。更新缓存的值时，请使用标准 放置 操作，否则可能会出现缓存损坏数据的可能性。

AdvancedCache advancedCache = cache.getAdvancedCache();

advancedCache.withFlags(Flag.CACHE_MODE_LOCAL, Flag.SKIP_LOCKING)
   .withFlags(Flag.FORCE_SYNCHRONOUS)
   .put("hello", "world");

Set<CompletableFuture<?>> futures = new HashSet<>();
futures.add(cache.putAsync(key1, value1)); // does not block
futures.add(cache.putAsync(key2, value2)); // does not block
futures.add(cache.putAsync(key3, value3)); // does not block

// the remote calls for the 3 puts will effectively be executed
// in parallel, particularly useful if running in distributed mode
// and the 3 keys would typically be pushed to 3 different nodes
// in the cluster

// check that the puts completed successfully
for (CompletableFuture<?> f: futures) f.get();

Data Grid 包括多个角色，为用户提供访问缓存和数据网格资源的权限。

CN=myapplication,OU=applications,DC=mycompany
CN=dataprocessors,OU=groups,DC=mycompany
CN=finance,OU=groups,DC=mycompany

dataprocessors
finance

dataprocessors: ALL_WRITE ALL_READ
finance: LISTEN

ALL_WRITE ALL_READ LISTEN

<cache-container>
  <security>
    <authorization>
      <common-name-role-mapper />
    </authorization>
  </security>
</cache-container>

{
  "infinispan" : {
    "cache-container" : {
      "security" : {
        "authorization" : {
          "common-name-role-mapper": {}
        }
      }
    }
  }
}

infinispan:
  cacheContainer:
    security:
      authorization:
        commonNameRoleMapper: ~

在使用嵌入式缓存时，您可以使用 GlobalSecurityConfigurationBuilder 和 ConfigurationBuilder 类配置授权。

将安全授权与数据网格缓存一起使用时，动态将角色映射到权限。

当您为使用安全授权的嵌入式缓存构建 DefaultCacheManager 时，Cache Manager 会返回 SecureCache，它会在调用任何操作前检查安全上下文。SecureCache 还确保应用程序无法检索较低级别的不安全对象，如 DataContainer。因此，您必须使用具有适当级别权限的 Data Grid 用户执行代码。

当您向用户授予或拒绝角色时，Data Grid 存储哪些用户可以在内部访问您的缓存的详细信息。此 ACL 缓存通过避免 Data Grid 的需求来计算每个请求执行读写操作的适当权限，从而提高了安全授权的性能。

ACL 缓存配置

<infinispan>
  <cache-container name="acl-cache-configuration">
    <security cache-size="1000"
              cache-timeout="300000">
      <authorization/>
    </security>
  </cache-container>
</infinispan>

{
  "infinispan" : {
    "cache-container" : {
      "name" : "acl-cache-configuration",
      "security" : {
        "cache-size" : "1000",
        "cache-timeout" : "300000",
        "authorization" : {}
      }
    }
  }
}

infinispan:
  cacheContainer:
    name: "acl-cache-configuration"
    security:
      cache-size: "1000"
      cache-timeout: "300000"
      authorization: ~

配置 Data Grid 以导出缓存管理器和嵌入式缓存的统计信息。

嵌入式缓存统计

<infinispan>
  <cache-container statistics="true">
    <distributed-cache statistics="true"/>
    <replicated-cache statistics="true"/>
  </cache-container>
</infinispan>

GlobalConfigurationBuilder global = GlobalConfigurationBuilder.defaultClusteredBuilder().cacheContainer().statistics(true);
DefaultCacheManager cacheManager = new DefaultCacheManager(global.build());

Configuration builder = new ConfigurationBuilder();
builder.statistics().enable();

Data Grid 生成与任何监控系统兼容的指标。

默认情况下，当启用统计数据时，Data Grid 会生成量表，但您也可以将其配置为生成直方图。

指标配置

<infinispan>
  <cache-container statistics="true">
    <metrics gauges="true"
             histograms="true" />
  </cache-container>
</infinispan>

{
  "infinispan" : {
    "cache-container" : {
      "statistics" : "true",
      "metrics" : {
        "gauges" : "true",
        "histograms" : "true"
      }
    }
  }
}

infinispan:
  cacheContainer:
    statistics: "true"
    metrics:
      gauges: "true"
      histograms: "true"

GlobalConfiguration globalConfig = new GlobalConfigurationBuilder()
  //Computes and collects statistics for the Cache Manager.
  .statistics().enable()
  //Exports collected statistics as gauge and histogram metrics.
  .metrics().gauges(true).histograms(true)
  .build();

Data Grid 可以注册可以用来收集统计信息并执行管理操作的 JMX MBeans。您还必须启用统计信息，否则 Data Grid 为 JMX MBeans 中的所有统计属性提供 0 值。

JMX 配置

<infinispan>
  <cache-container statistics="true">
    <jmx enabled="true"
         domain="example.com"/>
  </cache-container>
</infinispan>

{
  "infinispan" : {
    "cache-container" : {
      "statistics" : "true",
      "jmx" : {
        "enabled" : "true",
        "domain" : "example.com"
      }
    }
  }
}

infinispan:
  cacheContainer:
    statistics: "true"
    jmx:
      enabled: "true"
      domain: "example.com"

GlobalConfiguration global = GlobalConfigurationBuilder.defaultClusteredBuilder()
   .jmx().enable()
   .domain("org.mydomain");

<infinispan>
  <cache-container statistics="true">
    <jmx enabled="true"
         domain="example.com"
         mbean-server-lookup="com.example.MyMBeanServerLookup"/>
  </cache-container>
</infinispan>

{
  "infinispan" : {
    "cache-container" : {
      "statistics" : "true",
      "jmx" : {
        "enabled" : "true",
        "domain" : "example.com",
        "mbean-server-lookup" : "com.example.MyMBeanServerLookup"
      }
    }
  }
}

infinispan:
  cacheContainer:
    statistics: "true"
    jmx:
      enabled: "true"
      domain: "example.com"
      mbeanServerLookup: "com.example.MyMBeanServerLookup"

GlobalConfiguration global = GlobalConfigurationBuilder.defaultClusteredBuilder()
   .jmx().enable()
   .domain("org.mydomain")
   .mBeanServerLookup(new com.acme.MyMBeanServerLookup());

您可以为跨节点分布集群缓存导出时间指标。

当集群缓存拓扑更改时，状态传输操作发生，如加入或离开集群的节点。在状态传输操作期间，Data Grid 会导出每个缓存中的指标，以便您可以确定缓存的状态。状态传输将属性作为属性公开，以便 Data Grid 可以从每个缓存中导出指标。

Data Grid 生成与 REST API 和 JMX API 兼容的时间指标。

监控备份位置的站点状态，以检测站点之间的通信中断。当远程站点状态变为 离线时，Data Grid 会停止将数据复制到备份位置。您的数据不同步，您必须在使集群重新上线前修复不一致的问题。

对于早期问题检测，需要监控跨站点事件。使用以下监控策略之一：

使用 REST API 监控跨站点复制

使用 REST 端点监控所有缓存的跨站点复制状态。您可以实施自定义脚本来轮询 REST 端点，或者使用以下示例。

#!/usr/bin/python3
import time
import requests
from requests.auth import HTTPDigestAuth


class InfinispanConnection:

    def __init__(self, server: str = 'http://localhost:11222', cache_manager: str = 'default',
                 auth: tuple = ('admin', 'change_me')) -> None:
        super().__init__()
        self.__url = f'{server}/rest/v2/container/x-site/backups/'
        self.__auth = auth
        self.__headers = {
            'accept': 'application/json'
        }

    def get_sites_status(self):
        try:
            rsp = requests.get(self.__url, headers=self.__headers, auth=HTTPDigestAuth(self.__auth[0], self.__auth[1]))
            if rsp.status_code != 200:
                return None
            return rsp.json()
        except:
            return None


# Specify credentials for Data Grid user with permission to access the REST endpoint
USERNAME = 'admin'
PASSWORD = 'change_me'
# Set an interval between cross-site status checks
POLL_INTERVAL_SEC = 5
# Provide a list of servers
SERVERS = [
    InfinispanConnection('http://127.0.0.1:11222', auth=(USERNAME, PASSWORD)),
    InfinispanConnection('http://127.0.0.1:12222', auth=(USERNAME, PASSWORD))
]
#Specify the names of remote sites
REMOTE_SITES = [
    'nyc'
]
#Provide a list of caches to monitor
CACHES = [
    'work',
    'sessions'
]


def on_event(site: str, cache: str, old_status: str, new_status: str):
    # TODO implement your handling code here
    print(f'site={site} cache={cache} Status changed {old_status} -> {new_status}')


def __handle_mixed_state(state: dict, site: str, site_status: dict):
    if site not in state:
        state[site] = {c: 'online' if c in site_status['online'] else 'offline' for c in CACHES}
        return

    for cache in CACHES:
        __update_cache_state(state, site, cache, 'online' if cache in site_status['online'] else 'offline')


def __handle_online_or_offline_state(state: dict, site: str, new_status: str):
    if site not in state:
        state[site] = {c: new_status for c in CACHES}
        return

    for cache in CACHES:
        __update_cache_state(state, site, cache, new_status)


def __update_cache_state(state: dict, site: str, cache: str, new_status: str):
    old_status = state[site].get(cache)
    if old_status != new_status:
        on_event(site, cache, old_status, new_status)
        state[site][cache] = new_status


def update_state(state: dict):
    rsp = None
    for conn in SERVERS:
        rsp = conn.get_sites_status()
        if rsp:
            break
    if rsp is None:
        print('Unable to fetch site status from any server')
        return

    for site in REMOTE_SITES:
        site_status = rsp.get(site, {})
        new_status = site_status.get('status')
        if new_status == 'mixed':
            __handle_mixed_state(state, site, site_status)
        else:
            __handle_online_or_offline_state(state, site, new_status)


if __name__ == '__main__':
    _state = {}
    while True:
        update_state(_state)
        time.sleep(POLL_INTERVAL_SEC)

当站点状态从 Online 变为 offline 或 vice-versa 时，调用 on_event 的功能。

如果要使用这个脚本，您必须指定以下变量：

使用 Prometheus 指标监控跨站点复制

Prometheus 和其他监控系统可让您配置警报来检测站点状态何时更改为 离线。

Data Grid 在 infinispan-core-14.0.21.Final -redhat-00001.jar 文件的 default-configs 目录中提供默认的 JGroups 堆栈文件 default-jgroups114.xml。

Data Grid 支持不同的协议，允许节点在网络上自动找到并形成集群。

Data Grid 可以使用两种发现机制：

<PING num_discovery_runs="3"/>

<TCP bind_port="7800" />
<TCPPING timeout="3000"
         initial_hosts="${jgroups.tcpping.initial_hosts:hostname1[port1],hostname2[port2]}"
         port_range="0"
         num_initial_members="3"/>

<MPING mcast_addr="${jgroups.mcast_addr:239.6.7.8}"
       mcast_port="${jgroups.mcast_port:46655}"
       num_discovery_runs="3"
       ip_ttl="${jgroups.udp.ip_ttl:2}"/>

<TCP bind_port="7800" />
<TCPGOSSIP timeout="3000"
           initial_hosts="${GossipRouterAddress}"
           num_initial_members="3" />

<JDBC_PING connection_url="jdbc:mysql://localhost:3306/database_name"
           connection_username="user"
           connection_password="password"
           connection_driver="com.mysql.jdbc.Driver"/>

<dns.DNS_PING dns_query="myservice.myproject.svc.cluster.local" />

Data Grid 使用 JGroups 协议堆栈，以便节点可以在专用集群通道上发送其他消息。

Data Grid 为 UDP 和 TCP 协议提供预配置的 JGroups 堆栈。您可以使用这些默认堆栈作为构建自定义集群传输配置的起点，该配置根据您的网络要求进行了优化。

[org.infinispan.CLUSTER] ISPN000078: Starting JGroups channel cluster with stack udp

调整和调优属性，以创建适合您的网络要求的集群传输配置。

Data Grid 提供可让您扩展默认 JGroups 堆栈以方便配置的属性。您可以在合并、删除和替换其他属性时从默认堆栈继承属性。

在启动时将系统属性传递给 Data Grid，以调优群集传输。

例如，设置自定义绑定端口和 IP 地址，如下所示：

java -cp ... -Djgroups.bind.port=1234 -Djgroups.bind.address=192.0.2.0

您可以将完整的 JGroups 堆栈定义插入到 infinispan.xml 文件中。

引用在 infinispan.xml 文件中定义自定义 JGroups 堆栈的外部文件。

您还可以使用 TransportConfigurationBuilder 类中的 addProperty （） 方法来指定自定义 JGroups 堆栈文件，如下所示：

GlobalConfiguration globalConfig = new GlobalConfigurationBuilder().transport()
        .defaultTransport()
        .clusterName("prod-cluster")
        //Uses a custom JGroups stack for cluster transport.
        .addProperty("configurationFile", "my-jgroups-udp.xml")
        .build();

在本例中，my-jgroups-udp.xml 使用自定义属性引用 UDP 堆栈，如下所示：

<config xmlns="urn:org:jgroups"
        xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
        xsi:schemaLocation="urn:org:jgroups http://www.jgroups.org/schema/jgroups-4.2.xsd">
    <UDP bind_addr="${jgroups.bind_addr:127.0.0.1}"
         mcast_addr="${jgroups.udp.mcast_addr:239.0.2.0}"
         mcast_port="${jgroups.udp.mcast_port:46655}"
         tos="8"
         ucast_recv_buf_size="20000000"
         ucast_send_buf_size="640000"
         mcast_recv_buf_size="25000000"
         mcast_send_buf_size="640000"
         bundler.max_size="64000"
         ip_ttl="${jgroups.udp.ip_ttl:2}"
         diag.enabled="false"
         thread_naming_pattern="pl"
         thread_pool.enabled="true"
         thread_pool.min_threads="2"
         thread_pool.max_threads="30"
         thread_pool.keep_alive_time="5000" />
    <!-- Other JGroups stack configuration goes here. -->
</config>

构建自定义 JGroups JChannels，如下例所示：

GlobalConfigurationBuilder global = new GlobalConfigurationBuilder();
JChannel jchannel = new JChannel();
// Configure the jchannel as needed.
JGroupsTransport transport = new JGroupsTransport(jchannel);
global.transport().transport(transport);
new DefaultCacheManager(global.build());

保护集群传输，以便节点与加密消息通信。您还可以配置 Data Grid 集群来执行证书身份验证，以便只有具有有效身份的节点才能加入。

[org.infinispan.CLUSTER] ISPN000078: Starting JGroups channel cluster with stack <encrypted_stack_name>

[org.jgroups.protocols.ASYM_ENCRYPT] <hostname>: received message without encrypt header from <hostname>; dropping it

<infinispan>
  <jgroups>
    <!-- Creates a secure JGroups stack named "encrypt-tcp" that extends the default TCP stack. -->
    <stack name="encrypt-tcp" extends="tcp">
      <!-- Adds a keystore from which nodes obtain secret keys. -->
      <!-- Uses the stack.combine and stack.position attributes to insert SYM_ENCRYPT into the default TCP stack after VERIFY_SUSPECT2. -->
      <SYM_ENCRYPT keystore_name="myKeystore.p12"
                   keystore_type="PKCS12"
                   store_password="changeit"
                   key_password="changeit"
                   alias="myKey"
                   stack.combine="INSERT_AFTER"
                   stack.position="VERIFY_SUSPECT2"/>
    </stack>
  </jgroups>
  <cache-container name="default" statistics="true">
    <!-- Configures the cluster to use the JGroups stack. -->
    <transport cluster="${infinispan.cluster.name}"
               stack="encrypt-tcp"
               node-name="${infinispan.node.name:}"/>
  </cache-container>
</infinispan>

[org.infinispan.CLUSTER] ISPN000078: Starting JGroups channel cluster with stack <encrypted_stack_name>

[org.jgroups.protocols.SYM_ENCRYPT] <hostname>: received message without encrypt header from <hostname>; dropping it

Data Grid 对集群传输信息使用以下端口：

跨站点复制

Data Grid 为 JGroups RELAY2 协议使用以下端口：

Data Grid 提供了一个 ClusteredLock API，可让您在嵌入式模式中使用 Data Grid 时同时在集群中执行代码。

API 由以下内容组成：

如果为同一所有者发送两个连续锁定调用，则第一个调用会获取锁定（如果可用），第二个调用会被阻止。

了解如何在应用程序中嵌入的 Data Grid 中使用集群锁定。

<dependency>
   <groupId>org.infinispan</groupId>
   <artifactId>infinispan-clustered-lock</artifactId>
</dependency>

// Set up a clustered Cache Manager.
GlobalConfigurationBuilder global = GlobalConfigurationBuilder.defaultClusteredBuilder();

// Configure the cache mode, in this case it is distributed and synchronous.
ConfigurationBuilder builder = new ConfigurationBuilder();
builder.clustering().cacheMode(CacheMode.DIST_SYNC);

// Initialize a new default Cache Manager.
DefaultCacheManager cm = new DefaultCacheManager(global.build(), builder.build());

// Initialize a Clustered Lock Manager.
ClusteredLockManager clm1 = EmbeddedClusteredLockManagerFactory.from(cm);

// Define a clustered lock named 'lock'.
clm1.defineLock("lock");

// Get a lock from each node in the cluster.
ClusteredLock lock = clm1.get("lock");

AtomicInteger counter = new AtomicInteger(0);

// Acquire the lock as follows.
// Each 'lock.tryLock(1, TimeUnit.SECONDS)' method attempts to acquire the lock.
// If the lock is not available, the method waits for the timeout period to elapse. When the lock is acquired, other calls to acquire the lock are blocked until the lock is released.
CompletableFuture<Boolean> call1 = lock.tryLock(1, TimeUnit.SECONDS).whenComplete((r, ex) -> {
    if (r) {
        System.out.println("lock is acquired by the call 1");
        lock.unlock().whenComplete((nil, ex2) -> {
            System.out.println("lock is released by the call 1");
            counter.incrementAndGet();
        });
    }
});

CompletableFuture<Boolean> call2 = lock.tryLock(1, TimeUnit.SECONDS).whenComplete((r, ex) -> {
    if (r) {
        System.out.println("lock is acquired by the call 2");
        lock.unlock().whenComplete((nil, ex2) -> {
            System.out.println("lock is released by the call 2");
            counter.incrementAndGet();
        });
    }
});

CompletableFuture<Boolean> call3 = lock.tryLock(1, TimeUnit.SECONDS).whenComplete((r, ex) -> {
    if (r) {
        System.out.println("lock is acquired by the call 3");
        lock.unlock().whenComplete((nil, ex2) -> {
            System.out.println("lock is released by the call 3");
            counter.incrementAndGet();
        });
    }
});

CompletableFuture.allOf(call1, call2, call3).whenComplete((r, ex) -> {
    // Print the value of the counter.
    System.out.println("Value of the counter is " + counter.get());

    // Stop the Cache Manager.
    cm.stop();
});

集群锁定管理器包含一个存储锁定状态的内部缓存。您可以以声明性方式或以编程方式配置内部缓存。

由于您有一组计算机，因此最好利用其组合计算能力在所有这些计算机上执行代码。Cache Manager 附带一个 nice 工具，可让您在集群中执行任意代码。请注意，这个功能不需要使用缓存。此 Cluster Executor 可通过调用 EmbeddedCacheManager 上的 executor （）来检索。这个 executor 在集群和非集群配置中可以被检索。

此管理器专门围绕 Java 流 API 构建，因此所有方法都使用功能接口作为参数。因为这些参数将发送到其他节点，因此它们需要按顺序排序。我们甚至使用 nice trick 来确保我们的 lambda 可立即串行。这是通过参数实现 Serializable 和 real 参数类型（如 Runnable 或 Function）。在确定要调用的方法时，JRE 将选择最具体的类，因此在这样，您的 lambdas 始终可以被序列化。也可以使用外部工具来进一步减小消息大小。

默认情况下，管理器将向集群中的所有节点提交给定命令，包括从中提交它的节点。您可以使用 filterTargets 方法控制在其上执行任务的节点，如 部分中所述。

EmbeddedCacheManager manager = ...;
manager.executor().filterTargets(ClusterExecutionPolicy.SAME_RACK).submit(...)

EmbeddedCacheManager manager = ...;
// Just filter
manager.executor().filterTargets(a -> a.equals(..)).submit(...)
// Filter only those in the desired topology
manager.executor().filterTargets(ClusterExecutionPolicy.SAME_SITE, a -> a.equals(..)).submit(...)

EmbeddedCacheManager manager = ...;
manager.executor().singleNodeSubmission().submit(...)

public class PiAppx {

   public static void main (String [] arg){
      EmbeddedCacheManager cacheManager = ..
      boolean isCluster = ..

      int numPoints = 1_000_000_000;
      int numServers = isCluster ? cacheManager.getMembers().size() : 1;
      int numberPerWorker = numPoints / numServers;

      ClusterExecutor clusterExecutor = cacheManager.executor();
      long start = System.currentTimeMillis();
      // We receive results concurrently - need to handle that
      AtomicLong countCircle = new AtomicLong();
      CompletableFuture<Void> fut = clusterExecutor.submitConsumer(m -> {
         int insideCircleCount = 0;
         for (int i = 0; i < numberPerWorker; i++) {
            double x = Math.random();
            double y = Math.random();
            if (insideCircle(x, y))
               insideCircleCount++;
         }
         return insideCircleCount;
      }, (address, count, throwable) -> {
         if (throwable != null) {
            throwable.printStackTrace();
            System.out.println("Address: " + address + " encountered an error: " + throwable);
         } else {
            countCircle.getAndAdd(count);
         }
      });
      fut.whenComplete((v, t) -> {
         // This is invoked after all nodes have responded with a value or exception
         if (t != null) {
            t.printStackTrace();
            System.out.println("Exception encountered while waiting:" + t);
         } else {
            double appxPi = 4.0 * countCircle.get() / numPoints;

            System.out.println("Distributed PI appx is " + appxPi +
                  " using " + numServers + " node(s), completed in " + (System.currentTimeMillis() - start) + " ms");
         }
      });

      // May have to sleep here to keep alive if no user threads left
   }

   private static boolean insideCircle(double x, double y) {
      return (Math.pow(x - 0.5, 2) + Math.pow(y - 0.5, 2))
            <= Math.pow(0.5, 2);
   }
}

本节重点介绍了您正在使用的底层缓存类型的各种选项。

可以过滤流，使其仅在给定的密钥子集上运行。这可以通过在 CacheStream 上调用 filterKeys 方法来完成。这应该总是通过 Predicate 过滤器使用，如果 predicate 保存所有键，则更快。

如果您熟悉 AdvancedCache 接口，您可能会知道为什么在这个 keyFilter 上使用 getAll。如果您需要条目为 并在本地节点的内存中需要它们，则有一些小的好处（大较小的有效负载）来使用 getAll。但是，如果您需要在这些元素上处理流，因为您将得到分布式和线程并行处理。

这个选项只支持复制和分布式缓存。这允许用户在 KeyPartitioner 决定时对数据的子集进行操作。可以通过在 CacheStream 上调用 filterKeySegments 方法来过滤网段。这在键过滤器后应用，但在执行任何中间操作前应用。

与本地或无效缓存一起使用的流可以像在常规集合上使用流一样使用。如果需要，数据网格处理所有翻译，并与更有趣的选项（例如 storeAsBinary 和 cache loader）一起工作。仅执行流操作的节点本地数据，例如无效只使用本地条目。

以下代码使用一个缓存，并返回一个带有字符串 "JBoss" 的所有缓存条目的映射

Map<Object, String> jbossValues =
cache.entrySet().stream()
     .filter(e -> e.getValue().contains("JBoss"))
     .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));

这是流进入其范围的地方。执行流操作时，它会将各种中间和终端操作发送到具有相关数据的每个节点。这允许处理拥有数据的节点上的中间值，并且仅将最终结果发送回原始节点，从而提高性能。

Map<Object, String> jbossValues = cache.entrySet().stream()
              .filter(e -> e.getValue().contains("Jboss"))
              .collect(CacheCollectors.serializableCollector(() -> Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue)));

Map<Object, String> jbossValues = cache.entrySet().stream()
              .filter(e -> e.getValue().contains("Jboss"))
              .collect(() -> Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));

Map<Object, String> jbossValues = map.entrySet().stream()
              .filter((Serializable & Predicate<Map.Entry<Object, String>>) e -> e.getValue().contains("Jboss"))
              .collect(CacheCollectors.serializableCollector(() -> Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue)));

   Map<Object, String> jbossValues = cache.entrySet().stream()
              .filter(new ContainsFilter("Jboss"))
              .collect(() -> Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));

   class ContainsFilter implements Predicate<Map.Entry<Object, String>> {
      private final String target;

      ContainsFilter(String target) {
         this.target = target;
      }

      @Override
      public boolean test(Map.Entry<Object, String> e) {
         return e.getValue().contains(target);
      }
   }

   class JbossFilterExternalizer implements AdvancedExternalizer<ContainsFilter> {

      @Override
      public Set<Class<? extends ContainsFilter>> getTypeClasses() {
         return Util.asSet(ContainsFilter.class);
      }

      @Override
      public Integer getId() {
         return CUSTOM_ID;
      }

      @Override
      public void writeObject(ObjectOutput output, ContainsFilter object) throws IOException {
         output.writeUTF(object.target);
      }

      @Override
      public ContainsFilter readObject(ObjectInput input) throws IOException, ClassNotFoundException {
         return new ContainsFilter(input.readUTF());
      }
   }

Map<Object, String> map = (Map<Object, String>) cache.entrySet().stream()
              .filter(new ContainsFilter("Jboss"))
              .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));

 class ToMapCollectorSupplier<K, U> implements Supplier<Collector<Map.Entry<K, U>, ?, Map<K, U>>> {
      static final ToMapCollectorSupplier INSTANCE = new ToMapCollectorSupplier();

      private ToMapCollectorSupplier() { }

      @Override
      public Collector<Map.Entry<K, U>, ?, Map<K, U>> get() {
         return Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue);
      }
   }

   class ToMapCollectorSupplierExternalizer implements AdvancedExternalizer<ToMapCollectorSupplier> {

      @Override
      public Set<Class<? extends ToMapCollectorSupplier>> getTypeClasses() {
         return Util.asSet(ToMapCollectorSupplier.class);
      }

      @Override
      public Integer getId() {
         return CUSTOM_ID;
      }

      @Override
      public void writeObject(ObjectOutput output, ToMapCollectorSupplier object) throws IOException {
      }

      @Override
      public ToMapCollectorSupplier readObject(ObjectInput input) throws IOException, ClassNotFoundException {
         return ToMapCollectorSupplier.INSTANCE;
      }
   }

默认情况下，分布式流会尽量并行化。最终用户可以控制这一点，实际上它们始终需要控制其中一个选项。这些流的并行性有两种。

当从 缓存集合创建流时，最终用户可以在调用流或 parallelStream 方法之间进行选择。https://docs.oracle.com/javase/8/docs/api/java/util/Collection.html#stream--根据是否选择了并行流，将在本地为每个节点启用多个线程。请注意，一些操作（如重新哈希）和每个操作都会在本地使用顺序流。这在某些情况下可以增强，以允许本地并行流。

在使用本地并行时，用户应该小心，因为它需要大量条目或操作，这些条目或操作需要可以更快地计算。另请注意，如果用户使用具有 的并行流，并且每个操作不应阻断，这会在常见池中执行，这通常为计算操作保留。

当有多个节点时，远程请求可能需要控制同时处理远程请求是否同时处理。默认情况下，除 iterator 执行并发请求外的所有终端操作。迭代器，用于降低本地节点上的总体内存压力的方法，仅执行实际上性能稍好更高的顺序请求。

如果用户希望在 CacheStream 上调用 sequentialDistribution 或 parallelDistribution 方法来更改此默认值。

可以为操作请求设置超时值。这个超时仅用于远程请求超时，它基于每个请求。前者表示本地执行不会超时，后者意味着，如果您有一个故障转移场景，如后续请求上方描述，则每个请求都有新的超时。如果没有指定超时，它将使用复制超时作为默认超时时间。您可以通过执行以下操作在任务中设置超时：

CacheStream<Map.Entry<Object, String>> stream = cache.entrySet().stream();
stream.timeout(1, TimeUnit.MINUTES);

有关此问题的更多信息，请检查 java doc in timeout javadoc。

流 都有一个称为 的终端操作，允许对数据运行某种副作用的操作。https://docs.oracle.com/javase/8/docs/api/java/util/stream/Stream.html#forEach-java.util.function.Consumer-在这种情况下，可能需要获得支持此流的缓存的引用。如果您的 Consumer 实现 CacheAware 接口，则在来自 Consumer 接口的 accept 方法之前调用 injectCache 方法。

分布式流执行方式类似于映射减少。除了本例中，我们会向不同的节点发送零到多个中间操作（映射、过滤等）和单个终端操作。操作基本上会减少到以下几项：

在大多数情况下，所有操作都是完全分发的，因为操作是在每个远程节点上完全应用，通常只有最后一个操作或相关操作才可以重新应用，以减少多个节点的结果。务必要注意，中间值实际上不需要被序列化，它是发送的最后一个值，这是所需部分（突出显示各种操作的异常）。

终端 operator 分布的结果 会减少以下段落描述了各种终端操作器的分布式缩减如何工作。其中一些是特殊的，可能需要中间值才能被序列化，而不是最终结果。

迭代器、分割 器和每个 终端运算符都与重新哈希感知器不同的是，必须跟踪每个网段处理了哪些键，而不是仅段段。这是为了保证一次（迭代器和分割器）或至少一次行为（对于每个），即使在集群成员资格发生变化下也是如此。

在远程节点上调用时，迭代 器和 分割 器运算符将返回条目的回滚批处理，其中下一个批处理仅在最后一个完全被完全使用后发回。完成此批处理是为了限制给定时间在内存中有多少条目。用户节点将保存在其处理哪些密钥以及给定片段完成时，将从内存中释放这些密钥。这就是为什么在迭代器方法中使用顺序处理的原因，因此仅在内存中保存片段密钥的子集，而不是来自所有节点。

per （） 方法也会返回批处理，但在处理至少批处理后，它会返回批处理密钥。这样，原始节点可以知道已经处理了哪些密钥，以减少再次处理同一条目的几率。不幸的是，这意味着当节点意外停机时，至少有一次行为。在这种情况下，节点可能已经处理批处理，而还没有完成其中一个和这些条目，但还没有在完成的批处理中会在重新哈希失败时再次运行。请注意，在收到所有响应前，添加节点不会导致这个问题，因为重新哈希故障转移不会发生。

这些操作批处理的大小都由相同的值控制，这可以通过调用 CacheStream 上的 distributedBatchSize 方法进行配置。这个值将默认为在状态传输中配置的 chunkSize。不幸的是，这个值会权衡内存用量与性能，至少一次，您的鼠标可能会有所不同。

将 迭代器 与复制和分布式缓存一起使用

当节点是分布式流的所有请求片段的主要或备份所有者时，Data Grid 在本地执行 迭代 或 分割 器终端操作，这会优化性能，因为远程迭代更为资源密集型。

此优化适用于复制和分布式缓存。但是，在使用 共享 并启用了 write-behind 的缓存存储时，Data Grid 会远程执行迭代。在这种情况下，远程执行迭代可确保一致性。

有些带有特殊例外的中间操作 会跳过、peek、排序 1和 不同的。https://docs.oracle.com/javase/8/docs/api/java/util/stream/Stream.html#sorted--所有这些方法在流处理中都会发生某种形式，以保证正确性，如下所示。请注意，这意味着这些操作可能会导致严重的性能降级。

其余的中间操作会按预期完全分发。

字数

单词 count 是典型的，如果过度使用，如映射/缩减范例。假设我们在 Data Grid 节点上存储了 key → 句子的映射。Key 是一个字符串，每个句子都是一个 String，我们必须计算所有句子中所有可用的词语。此类分布式任务的实现可定义如下：

public class WordCountExample {

   /**
    * In this example replace c1 and c2 with
    * real Cache references
    *
    * @param args
    */
   public static void main(String[] args) {
      Cache<String, String> c1 = ...;
      Cache<String, String> c2 = ...;

      c1.put("1", "Hello world here I am");
      c2.put("2", "Infinispan rules the world");
      c1.put("3", "JUDCon is in Boston");
      c2.put("4", "JBoss World is in Boston as well");
      c1.put("12","JBoss Application Server");
      c2.put("15", "Hello world");
      c1.put("14", "Infinispan community");
      c2.put("15", "Hello world");

      c1.put("111", "Infinispan open source");
      c2.put("112", "Boston is close to Toronto");
      c1.put("113", "Toronto is a capital of Ontario");
      c2.put("114", "JUDCon is cool");
      c1.put("211", "JBoss World is awesome");
      c2.put("212", "JBoss rules");
      c1.put("213", "JBoss division of RedHat ");
      c2.put("214", "RedHat community");

      Map<String, Long> wordCountMap = c1.entrySet().parallelStream()
         .map(e -> e.getValue().split("\\s"))
         .flatMap(Arrays::stream)
         .collect(() -> Collectors.groupingBy(Function.identity(), Collectors.counting()));
   }
}

在这种情况下，可以简单地执行上例中的单词 count。

但是，如果我们想要在示例中找到最频繁的字词，该怎么办？如果您考虑第二个情况，您将意识到您需要首先计算所有词语，并首先可用。因此，我们实际上有一些选项。

我们可以在收集器上使用一个 finisher，它会在收集所有结果后在用户线程上调用。之前示例中删除了一些冗余行。

public class WordCountExample {
   public static void main(String[] args) {
      // Lines removed

      String mostFrequentWord = c1.entrySet().parallelStream()
         .map(e -> e.getValue().split("\\s"))
         .flatMap(Arrays::stream)
         .collect(() -> Collectors.collectingAndThen(
            Collectors.groupingBy(Function.identity(), Collectors.counting()),
               wordCountMap -> {
                  String mostFrequent = null;
                  long maxCount = 0;
                     for (Map.Entry<String, Long> e : wordCountMap.entrySet()) {
                        int count = e.getValue().intValue();
                        if (count > maxCount) {
                           maxCount = count;
                           mostFrequent = e.getKey();
                        }
                     }
                     return mostFrequent;
               }));

}

不幸的是，最后一步只会在单一线程中运行，如果我们有很多词语可能非常慢。可以通过另一种方式与 Streams 并行化。

在处理后，我们提到了本地节点，因此我们实际上可以在映射结果上使用流。因此，我们可以在结果中使用并行流。

public class WordFrequencyExample {
   public static void main(String[] args) {
      // Lines removed

      Map<String, Long> wordCount = c1.entrySet().parallelStream()
              .map(e -> e.getValue().split("\\s"))
              .flatMap(Arrays::stream)
              .collect(() -> Collectors.groupingBy(Function.identity(), Collectors.counting()));
      Optional<Map.Entry<String, Long>> mostFrequent = wordCount.entrySet().parallelStream().reduce(
              (e1, e2) -> e1.getValue() > e2.getValue() ? e1 : e2);

这样，在计算最频繁的元素时，您仍然可以在本地使用所有内核。

删除特定条目

分布式流也可以用作修改它所在数据的方法。例如，您可能想要删除包含特定词语的缓存中的所有条目。

public class RemoveBadWords {
   public static void main(String[] args) {
      // Lines removed
      String word = ..

      c1.entrySet().parallelStream()
         .filter(e -> e.getValue().contains(word))
         .forEach((c, e) -> c.remove(e.getKey()));

如果我们仔细注意了要序列化的内容，并且没有什么，我们注意到，只有词语以及操作才会按顺序化到其他 nods，因为它由 lambda 捕获。但是，实际保存部分是缓存操作在主所有者上执行，从而减少从缓存中删除这些值所需的网络流量数量。lambda 不会捕获缓存，因为我们在每个节点上调用时提供一个特殊的 BiConsumer 方法覆盖，这会将缓存传递给 BiConsumer

以这种方式考虑使用 for each 命令的一个问题是底层流没有获得锁定。缓存删除操作仍将自然而获得锁定，但该值可能会从流看到的内容改变。这意味着，在流读取后条目可能会改变，但删除会实际删除它。

我们特别添加了名为 LockedStream 的新变体。

其他示例的 plenty

Streams API 是一个 JRE 工具，有许多示例使用它。只需记住您的操作需要以某种方式 Serializ。

要在应用程序中使用 Red Hat JBoss EAP 提供的 Data Grid 库，请在应用程序的 pom.xml 文件中添加 Data Grid 依赖项。

在红帽 JBoss EAP 中创建数据网格缓存。

您可以通过资源查找来访问应用程序中的 Data Grid 缓存。

使用以下依赖项之一更新 pom.xml，以在项目中包含 Data Grid CDI 扩展：

<dependency>
  <groupId>org.infinispan</groupId>
  <artifactId>infinispan-cdi-embedded</artifactId>
</dependency>

<dependency>
  <groupId>org.infinispan</groupId>
  <artifactId>infinispan-cdi-remote</artifactId>
</dependency>

设置 CDI Bean 以注入嵌入式缓存。

设置 CDI Bean 以注入远程缓存。

您可以使用 CDI 事件接收缓存和缓存管理器级别事件。

import jakarta.transaction.event.Observes;
import org.infinispan.notifications.cachemanagerlistener.event.CacheStartedEvent;
import org.infinispan.notifications.cachelistener.event.*;

public class GreetingService {

    // Cache level events
    private void entryRemovedFromCache(@Observes CacheEntryCreatedEvent event) {
        ...
    }

    // Cache manager level events
    private void cacheStarted(@Observes CacheStartedEvent event) {
        ...
    }
}

MutimapCache 是一种 Data Grid Cache 类型，用于将键映射到每个键可以包含多个值的值。

<dependency>
  <groupId>org.infinispan</groupId>
  <artifactId>infinispan-multimap</artifactId>
</dependency>

public interface MultimapCache<K, V> {

   CompletableFuture<Optional<CacheEntry<K, Collection<V>>>> getEntry(K key);

   CompletableFuture<Void> remove(SerializablePredicate<? super V> p);

   CompletableFuture<Void> put(K key, V value);

   CompletableFuture<Collection<V>> get(K key);

   CompletableFuture<Boolean> remove(K key);

   CompletableFuture<Boolean> remove(K key, V value);

   CompletableFuture<Void> remove(Predicate<? super V> p);

   CompletableFuture<Boolean> containsKey(K key);

   CompletableFuture<Boolean> containsValue(V value);

   CompletableFuture<Boolean> containsEntry(K key, V value);

   CompletableFuture<Long> size();

   boolean supportsDuplicates();

}

MultimapCache<String, String> multimapCache = ...;

multimapCache.put("girlNames", "marie")
             .thenCompose(r1 -> multimapCache.put("girlNames", "oihana"))
             .thenCompose(r3 -> multimapCache.get("girlNames"))
             .thenAccept(names -> {
                          if(names.contains("marie"))
                              System.out.println("Marie is a girl name");

                           if(names.contains("oihana"))
                              System.out.println("Oihana is a girl name");
                        });

Marie is a girl name
Oihana is a girl name

// create or obtain your EmbeddedCacheManager
EmbeddedCacheManager cm = ... ;

// create or obtain a MultimapCacheManager passing the EmbeddedCacheManager
MultimapCacheManager multimapCacheManager = EmbeddedMultimapCacheManagerFactory.from(cm);

// define the configuration for the multimap cache
multimapCacheManager.defineConfiguration(multimapCacheName, c.build());

// get the multimap cache
multimapCache = multimapCacheManager.get(multimapCacheName);