此版本仍在开发中，尚不被认为是稳定的。对于最新的稳定版本，请使用 Spring Integration 6.5.1！spring-doc.cadn.net.cn

Hazelcast 支持

Spring Integration提供了通道适配器和其他实用程序组件，以与内存中数据网格Hazelcast进行交互。spring-doc.cadn.net.cn

您需要将此依赖项包含在您的项目中：spring-doc.cadn.net.cn

<dependency>
    <groupId>org.springframework.integration</groupId>
    <artifactId>spring-integration-hazelcast</artifactId>
    <version>6.3.12-SNAPSHOT</version>
</dependency>
compile "org.springframework.integration:spring-integration-hazelcast:6.3.12-SNAPSHOT"

Hazelcast 组件的 XML 命名空间和 schemaLocation 定义包括：spring-doc.cadn.net.cn

xmlns:int-hazelcast="http://www.springframework.org/schema/integration/hazelcast"
xsi:schemaLocation="http://www.springframework.org/schema/integration/hazelcast
          https://www.springframework.org/schema/integration/hazelcast/spring-integration-hazelcast.xsd"

Hazelcast 事件驱动的入站通道适配器

Hazelcast 提供分布式数据结构，例如：spring-doc.cadn.net.cn

它还提供事件侦听器，以便侦听对这些数据结构所做的修改。spring-doc.cadn.net.cn

Hazelcast 事件驱动入站通道适配器侦听相关缓存事件并将事件消息发送到定义的通道。 它支持 XML 和 JavaConfig 驱动的配置。spring-doc.cadn.net.cn

XML 配置：

<int-hazelcast:inbound-channel-adapter channel="mapChannel"
                      cache="map"
                      cache-events="UPDATED, REMOVED"
                      cache-listening-policy="SINGLE" />

Hazelcast 事件驱动的入站通道适配器需要以下属性：spring-doc.cadn.net.cn

  • channel：指定将消息发送到的通道;spring-doc.cadn.net.cn

  • cache：指定要侦听的分布式对象引用。 这是一个强制性属性;spring-doc.cadn.net.cn

  • cache-events：指定要侦听的缓存事件。 它是一个可选属性，其默认值为ADDED. 其支持的值如下：spring-doc.cadn.net.cn

  • 支持的缓存事件类型IMap和MultiMap:ADDED,REMOVED,UPDATED,EVICTED,EVICT_ALL和CLEAR_ALL;spring-doc.cadn.net.cn

  • 支持的缓存事件类型ReplicatedMap:ADDED,REMOVED,UPDATED,EVICTED;spring-doc.cadn.net.cn

  • 支持的缓存事件类型IList,ISet和IQueue:ADDED,REMOVED. 没有缓存事件类型ITopic.spring-doc.cadn.net.cn

  • cache-listening-policy：将缓存侦听策略指定为SINGLE或ALL. 它是一个可选属性，其默认值为SINGLE. 每个 Hazelcast 入站通道适配器侦听具有相同 cache-events 属性的同一缓存对象，可以接收单个事件消息或所有事件消息。 如果是ALL，所有侦听具有相同 cache-events 属性的同一缓存对象的 Hazelcast 入站通道适配器都将接收所有事件消息。 如果是SINGLE，他们将收到独特的事件消息。spring-doc.cadn.net.cn

一些配置示例：spring-doc.cadn.net.cn

分布式地图
<int:channel id="mapChannel"/>

<int-hazelcast:inbound-channel-adapter channel="mapChannel"
                              cache="map"
                              cache-events="UPDATED, REMOVED" />

<bean id="map" factory-bean="instance" factory-method="getMap">
    <constructor-arg value="map"/>
</bean>

<bean id="instance" class="com.hazelcast.core.Hazelcast"
            factory-method="newHazelcastInstance">
    <constructor-arg>
        <bean class="com.hazelcast.config.Config" />
    </constructor-arg>
</bean>
分布式多地图
<int-hazelcast:inbound-channel-adapter channel="multiMapChannel"
                              cache="multiMap"
                              cache-events="ADDED, REMOVED, CLEAR_ALL" />

<bean id="multiMap" factory-bean="instance" factory-method="getMultiMap">
    <constructor-arg value="multiMap"/>
</bean>
分布列表
<int-hazelcast:inbound-channel-adapter  channel="listChannel"
                               cache="list"
                               cache-events="ADDED, REMOVED"
                               cache-listening-policy="ALL" />

<bean id="list" factory-bean="instance" factory-method="getList">
    <constructor-arg value="list"/>
</bean>
分布式集
<int-hazelcast:inbound-channel-adapter channel="setChannel" cache="set" />

<bean id="set" factory-bean="instance" factory-method="getSet">
    <constructor-arg value="set"/>
</bean>
分布式队列
<int-hazelcast:inbound-channel-adapter  channel="queueChannel"
                               cache="queue"
                               cache-events="REMOVED"
                               cache-listening-policy="ALL" />

<bean id="queue" factory-bean="instance" factory-method="getQueue">
    <constructor-arg value="queue"/>
</bean>
分布式主题
<int-hazelcast:inbound-channel-adapter channel="topicChannel" cache="topic" />

<bean id="topic" factory-bean="instance" factory-method="getTopic">
    <constructor-arg value="topic"/>
</bean>
复制的地图
<int-hazelcast:inbound-channel-adapter channel="replicatedMapChannel"
                              cache="replicatedMap"
                              cache-events="ADDED, UPDATED, REMOVED"
                              cache-listening-policy="SINGLE"  />

<bean id="replicatedMap" factory-bean="instance" factory-method="getReplicatedMap">
    <constructor-arg value="replicatedMap"/>
</bean>

Java 配置示例：

以下示例显示了DistributedMap配置。 相同的配置可用于其他分布式数据结构（IMap,MultiMap,ReplicatedMap,IList,ISet,IQueue和ITopic):spring-doc.cadn.net.cn

@Bean
public PollableChannel distributedMapChannel() {
    return new QueueChannel();
}

@Bean
public IMap<Integer, String> distributedMap() {
    return hazelcastInstance().getMap("Distributed_Map");
}

@Bean
public HazelcastInstance hazelcastInstance() {
    return Hazelcast.newHazelcastInstance();
}

@Bean
public HazelcastEventDrivenMessageProducer hazelcastEventDrivenMessageProducer() {
    final HazelcastEventDrivenMessageProducer producer = new HazelcastEventDrivenMessageProducer(distributedMap());
    producer.setOutputChannel(distributedMapChannel());
    producer.setCacheEventTypes("ADDED,REMOVED,UPDATED,CLEAR_ALL");
    producer.setCacheListeningPolicy(CacheListeningPolicyType.SINGLE);

    return producer;
}

Hazelcast 连续查询入站通道适配器

Hazelcast 连续查询允许侦听对特定映射条目执行的修改。Hazelcast 连续查询入站通道适配器是一个事件驱动的通道适配器，它根据定义的谓词侦听相关的分布式映射事件。spring-doc.cadn.net.cn

@Bean
public PollableChannel cqDistributedMapChannel() {
    return new QueueChannel();
}

@Bean
public IMap<Integer, String> cqDistributedMap() {
    return hazelcastInstance().getMap("CQ_Distributed_Map");
}

@Bean
public HazelcastInstance hazelcastInstance() {
    return Hazelcast.newHazelcastInstance();
}

@Bean
public HazelcastContinuousQueryMessageProducer hazelcastContinuousQueryMessageProducer() {
    final HazelcastContinuousQueryMessageProducer producer =
        new HazelcastContinuousQueryMessageProducer(cqDistributedMap(), "surname=TestSurname");
    producer.setOutputChannel(cqDistributedMapChannel());
    producer.setCacheEventTypes("UPDATED");
    producer.setIncludeValue(false);

    return producer;
}
<int:channel id="cqMapChannel"/>

<int-hazelcast:cq-inbound-channel-adapter
                channel="cqMapChannel"
                cache="cqMap"
                cache-events="UPDATED, REMOVED"
                predicate="name=TestName AND surname=TestSurname"
                include-value="true"
                cache-listening-policy="SINGLE"/>

<bean id="cqMap" factory-bean="instance" factory-method="getMap">
    <constructor-arg value="cqMap"/>
</bean>

<bean id="instance" class="com.hazelcast.core.Hazelcast"
            factory-method="newHazelcastInstance">
    <constructor-arg>
        <bean class="com.hazelcast.config.Config" />
    </constructor-arg>
</bean>

它支持以下六个属性：spring-doc.cadn.net.cn

  • channel：指定将消息发送到的通道;spring-doc.cadn.net.cn

  • cache：指定要侦听的分布式地图引用。 命令的;spring-doc.cadn.net.cn

  • cache-events：指定要侦听的缓存事件。带有ADDED作为其默认值。支持的值包括ADDED,REMOVED,UPDATED,EVICTED,EVICT_ALL和CLEAR_ALL;spring-doc.cadn.net.cn

  • predicate：指定一个谓词来侦听对特定映射条目执行的修改。 命令的;spring-doc.cadn.net.cn

  • include-value：指定在连续查询结果中包含 value 和 oldValue。可选true为默认;spring-doc.cadn.net.cn

  • cache-listening-policy：将缓存侦听策略指定为SINGLE或ALL. 可选，默认值为SINGLE. 每个 Hazelcast CQ 入站通道适配器侦听具有相同 cache-events 属性的同一缓存对象，可以接收单个事件消息或所有事件消息。如果是ALL，则侦听具有相同 cache-events 属性的同一缓存对象的所有 Hazelcast CQ 入站通道适配器都将收到所有事件消息。如果是SINGLE，他们将收到独特的事件消息。spring-doc.cadn.net.cn

Hazelcast 集群监视器入站通道适配器

Hazelcast 群集监视器支持侦听对群集执行的修改。Hazelcast 群集监视器入站通道适配器是一个事件驱动的通道适配器，可侦听相关的成员身份、分布式对象、迁移、生命周期和客户端事件：spring-doc.cadn.net.cn

@Bean
public PollableChannel eventChannel() {
    return new QueueChannel();
}

@Bean
public HazelcastInstance hazelcastInstance() {
    return Hazelcast.newHazelcastInstance();
}

@Bean
public HazelcastClusterMonitorMessageProducer hazelcastClusterMonitorMessageProducer() {
    HazelcastClusterMonitorMessageProducer producer = new HazelcastClusterMonitorMessageProducer(hazelcastInstance());
    producer.setOutputChannel(eventChannel());
    producer.setMonitorEventTypes("DISTRIBUTED_OBJECT");

    return producer;
}
<int:channel id="monitorChannel"/>

<int-hazelcast:cm-inbound-channel-adapter
                 channel="monitorChannel"
                 hazelcast-instance="instance"
                 monitor-types="MEMBERSHIP, DISTRIBUTED_OBJECT" />

<bean id="instance" class="com.hazelcast.core.Hazelcast"
            factory-method="newHazelcastInstance">
    <constructor-arg>
        <bean class="com.hazelcast.config.Config" />
    </constructor-arg>
</bean>

它支持以下三个属性：spring-doc.cadn.net.cn

  • channel：指定将消息发送到的通道;spring-doc.cadn.net.cn

  • hazelcast-instance：指定用于侦听集群事件的 Hazelcast 实例引用。它是必填属性;spring-doc.cadn.net.cn

  • monitor-types：指定要侦听的监视器类型。它是一个可选属性，具有MEMBERSHIP为默认值。支持的值为MEMBERSHIP,DISTRIBUTED_OBJECT,MIGRATION,LIFECYCLE,CLIENT.spring-doc.cadn.net.cn

Hazelcast 分布式 SQL 入站通道适配器

Hazelcast 允许在分布式映射上运行分布式查询。Hazelcast 分布式 SQL 入站通道适配器是一个轮询入站通道适配器。它运行定义的 distributed-sql 命令，并根据迭代类型返回结果。spring-doc.cadn.net.cn

@Bean
public PollableChannel dsDistributedMapChannel() {
    return new QueueChannel();
}

@Bean
public IMap<Integer, String> dsDistributedMap() {
    return hazelcastInstance().getMap("DS_Distributed_Map");
}

@Bean
public HazelcastInstance hazelcastInstance() {
    return Hazelcast.newHazelcastInstance();
}

@Bean
@InboundChannelAdapter(value = "dsDistributedMapChannel", poller = @Poller(maxMessagesPerPoll = "1"))
public HazelcastDistributedSQLMessageSource hazelcastDistributedSQLMessageSource() {
    final HazelcastDistributedSQLMessageSource messageSource =
        new HazelcastDistributedSQLMessageSource(dsDistributedMap(),
            "name='TestName' AND surname='TestSurname'");
    messageSource.setIterationType(DistributedSQLIterationType.ENTRY);

    return messageSource;
}
<int:channel id="dsMapChannel"/>

<int-hazelcast:ds-inbound-channel-adapter
            channel="dsMapChannel"
            cache="dsMap"
            iteration-type="ENTRY"
            distributed-sql="active=false OR age >= 25 OR name = 'TestName'">
    <int:poller fixed-delay="100"/>
</int-hazelcast:ds-inbound-channel-adapter>

<bean id="dsMap" factory-bean="instance" factory-method="getMap">
    <constructor-arg value="dsMap"/>
</bean>

<bean id="instance" class="com.hazelcast.core.Hazelcast"
            factory-method="newHazelcastInstance">
    <constructor-arg>
        <bean class="com.hazelcast.config.Config" />
    </constructor-arg>
</bean>

它需要轮询器并支持四个属性：spring-doc.cadn.net.cn

  • channel：指定将消息发送到的通道。 这是一个强制性属性;spring-doc.cadn.net.cn

  • cache：指定分布式IMap引用。 它是强制性属性;spring-doc.cadn.net.cn

  • iteration-type：指定结果类型。分布式 SQL 可以在EntrySet,KeySet,LocalKeySet或Values. 它是一个可选属性，具有VALUE为默认值。支持的值为ENTRY, `KEY,LOCAL_KEY和VALUE;spring-doc.cadn.net.cn

  • distributed-sql：指定 sql 语句的 where 子句。它是必填属性。spring-doc.cadn.net.cn

Hazelcast 出站通道适配器

Hazelcast 出站通道适配器侦听其定义的通道，并将传入消息写入相关的分布式缓存。它需要以下一个cache,cache-expression或HazelcastHeaders.CACHE_NAME用于分布式对象定义。支持的分布式对象包括：IMap,MultiMap,ReplicatedMap,IList,ISet,IQueue和ITopic.spring-doc.cadn.net.cn

@Bean
public MessageChannel distributedMapChannel() {
    return new DirectChannel();
}

@Bean
public IMap<Integer, String> distributedMap() {
    return hzInstance().getMap("Distributed_Map");
}

@Bean
public HazelcastInstance hzInstance() {
    return Hazelcast.newHazelcastInstance();
}

@Bean
@ServiceActivator(inputChannel = "distributedMapChannel")
public HazelcastCacheWritingMessageHandler hazelcastCacheWritingMessageHandler() {
    HazelcastCacheWritingMessageHandler handler = new HazelcastCacheWritingMessageHandler();
    handler.setDistributedObject(distributedMap());
    handler.setKeyExpression(new SpelExpressionParser().parseExpression("payload.id"));
    handler.setExtractPayload(true);
    return handler;
}
<int-hazelcast:outbound-channel-adapter channel="mapChannel"
                    cache-expression="headers['CACHE_HEADER']"
                    key-expression="payload.key"
                    extract-payload="true"/>

它需要以下属性：spring-doc.cadn.net.cn

  • channel：指定将消息发送到的通道;spring-doc.cadn.net.cn

  • cache：指定分布式对象引用。 自选;spring-doc.cadn.net.cn

  • cache-expression：通过 Spring 表达式语言 （SpEL） 指定分布式对象。 自选;spring-doc.cadn.net.cn

  • key-expression：通过 Spring 表达式语言 （SpEL） 指定键值对的键。 可选且仅对IMap,MultiMap和ReplicatedMap分布式数据结构。spring-doc.cadn.net.cn

  • extract-payload：指定是发送整个消息还是仅发送有效负载。 可选属性true是默认值。 如果为 true，则仅将有效负载写入分布式对象。 否则，将通过转换消息头和有效负载来写入整个消息。spring-doc.cadn.net.cn

通过在标头中设置分布式对象名称，可以通过同一通道将消息写入不同的分布式对象。 如果cache或cache-expression属性未定义，则HazelcastHeaders.CACHE_NAME必须在请求中设置标头Message.spring-doc.cadn.net.cn

Hazelcast 领导人选举

如果需要选举领导者（例如，对于只有一个节点应该接收消息的高可用性消息消费者），则基于 HazelcastLeaderInitiator可以使用：spring-doc.cadn.net.cn

@Bean
public HazelcastInstance hazelcastInstance() {
    return Hazelcast.newHazelcastInstance();
}

@Bean
public LeaderInitiator initiator() {
    return new LeaderInitiator(hazelcastInstance());
}

当一个节点被选为领导者时，它将发送一个OnGrantedEvent到所有应用程序侦听器。spring-doc.cadn.net.cn

Hazelcast 消息商店

对于分布式消息传递状态管理，例如对于持久性QueueChannel或跟踪Aggregator消息组，HazelcastMessageStore提供实现：spring-doc.cadn.net.cn

@Bean
public HazelcastInstance hazelcastInstance() {
    return Hazelcast.newHazelcastInstance();
}

@Bean
public MessageGroupStore messageStore() {
    return new HazelcastMessageStore(hazelcastInstance());
}

默认情况下，SPRING_INTEGRATION_MESSAGE_STORE IMap用于将消息和组存储为键/值。 任何自定义IMap可以提供给HazelcastMessageStore.spring-doc.cadn.net.cn

Hazelcast 元数据存储

实现ListenableMetadataStore可使用后备 HazelcastIMap. 默认地图是使用名称创建的SPRING_INTEGRATION_METADATA_STORE可以定制。spring-doc.cadn.net.cn

@Bean
public HazelcastInstance hazelcastInstance() {
    return Hazelcast.newHazelcastInstance();
}

@Bean
public MetadataStore metadataStore() {
    return new HazelcastMetadataStore(hazelcastInstance());
}

这HazelcastMetadataStore实现ListenableMetadataStore它允许您注册自己的MetadataStoreListener通过addListener(MetadataStoreListener callback).spring-doc.cadn.net.cn

Hazelcast 锁注册表

实现LockRegistry使用分发的后备 Hazelcast 可用ILock支持：spring-doc.cadn.net.cn

@Bean
public HazelcastInstance hazelcastInstance() {
    return Hazelcast.newHazelcastInstance();
}

@Bean
public LockRegistry lockRegistry() {
    return new HazelcastLockRegistry(hazelcastInstance());
}

与共享的MessageGroupStore（例如Aggregatorstore 管理）、HazelcastLockRegistry可用于跨多个应用程序实例提供此功能，以便一次只有一个实例可以作该组。spring-doc.cadn.net.cn

对于所有分布式作，必须在HazelcastInstance.

使用 Hazelcast 的消息通道

榛子IQueue和ITopic分布式对象本质上是消息传递原语，可以与 Spring Integration 核心组件一起使用，而无需在此 Hazelcast 模块中进行额外的实现。spring-doc.cadn.net.cn

这QueueChannel可由任何java.util.Queue，包括提到的 Hazelcast 分发IQueue:spring-doc.cadn.net.cn

@Bean
PollableChannel hazelcastQueueChannel(HazelcastInstance hazelcastInstance) {
    return new QueueChannel(hazelcastInstance.getQueue("springIntegrationQueue"));
}

将此配置放在应用程序的 Hazelcast 集群中的多个节点上，将使QueueChannel作为分布式，并且只有一个节点能够轮询单个Message由此IQueue. 其工作原理类似于PollableJmsChannel,PollableKafkaChannel或PollableAmqpChannel.spring-doc.cadn.net.cn

如果生产者端不是 Spring Integration 应用程序，则无法配置QueueChannel，因此是普通的榛子IQueueAPI 用于生成数据。 在这种情况下，QueueChannel方法在使用者端是错误的：必须改用入站通道适配器解决方案：spring-doc.cadn.net.cn

@Bean
public IQueue<String> myStringHzQueue(HazelcastInstance hazelcastInstance) {
    return hazelcastInstance.getQueue("springIntegrationQueue");
}

@Bean
@InboundChannelAdapter(channel = "stringValuesFromHzQueueChannel")
Supplier<String> fromHzIQueueSource(IQueue<String> myStringHzQueue) {
    return myStringHzQueue::poll;
}

这ITopicHazelcast 中的抽象具有与Topic在 JMS 中：所有订阅者都会收到已发布的消息。 搭配一双简单的MessageChannelbean 此机制作为开箱即用功能得到支持：spring-doc.cadn.net.cn

@Bean
public ITopic<Message<?>> springIntegrationTopic(HazelcastInstance hazelcastInstance,
        MessageChannel fromHazelcastTopicChannel) {

    ITopic<Message<?>> topic = hazelcastInstance.getTopic("springIntegrationTopic");
	topic.addMessageListener(m -> fromHazelcastTopicChannel.send(m.getMessageObject()));
	return topic;
}

@Bean
public MessageChannel publishToHazelcastTopicChannel(ITopic<Message<?>> springIntegrationTopic) {
    return new FixedSubscriberChannel(springIntegrationTopic::publish);
}

@Bean
public MessageChannel fromHazelcastTopicChannel() {
    return new DirectChannel();
}

这FixedSubscriberChannel是DirectChannel，这需要一个MessageHandler初始化时。 由于MessageHandler是一个函数接口，一个简单的 lambda，用于handleMessage方法可以提供。 当消息发送到publishToHazelcastTopicChannel它刚刚发布在 Hazelcast 上ITopic. 这com.hazelcast.topic.MessageListener也是一个函数式接口，因此 lambda 的ITopic#addMessageListener可以提供。 因此，订阅者fromHazelcastTopicChannel将消耗发送到上述ITopic.spring-doc.cadn.net.cn

一ExecutorChannel可配备IExecutorService. 例如，通过相应的配置，可以实现集群范围的单例：spring-doc.cadn.net.cn

@Bean
public HazelcastInstance hazelcastInstance() {
    return Hazelcast.newHazelcastInstance(
                new Config()
                    .addExecutorConfig(new ExecutorConfig()
                         .setName("singletonExecutor")
                         .setPoolSize(1)));
}

@Bean
public MessageChannel hazelcastSingletonExecutorChannel(HazelcastInstance hazelcastInstance) {
    return new ExecutorChannel(hazelcastInstance.getExecutorService("singletonExecutor"));
}