Hazelcast Support

Spring Integration 提供通道适配器和其他实用组件来与内存数据网格 Hazelcast 交互。

Spring Integration provides channel adapters and other utility components to interact with an in-memory data grid Hazelcast.

你需要将此依赖项包含在你的项目中：

You need to include this dependency into your project:

  • Maven

  • Gradle

<dependency>
    <groupId>org.springframework.integration</groupId>
    <artifactId>spring-integration-hazelcast</artifactId>
    <version>{project-version}</version>
</dependency>
compile "org.springframework.integration:spring-integration-hazelcast:{project-version}"

Hazelcast 组件的 XML 命名空间和 schemaLocation 定义为：

The XML namespace and schemaLocation definitions for Hazelcast components are:

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 Event-driven Inbound Channel Adapter

Hazelcast 提供分布式数据结构，例如：

Hazelcast provides distributed data structures such as:

  • com.hazelcast.map.IMap

  • com.hazelcast.multimap.MultiMap

  • com.hazelcast.collection.IList

  • com.hazelcast.collection.ISet

  • com.hazelcast.collection.IQueue

  • com.hazelcast.topic.ITopic

  • com.hazelcast.replicatedmap.ReplicatedMap

它还提供事件侦听器，以便侦听对这些数据结构所做的修改。

It also provides event listeners in order to listen to modifications made to these data structures.

  • com.hazelcast.core.EntryListener<K, V>

  • com.hazelcast.collection.ItemListener

  • com.hazelcast.topic.MessageListener

Hazelcast 事件驱动的入站通道适配器侦听相关的缓存事件，并将事件消息发送到定义的通道。它同时支持 XML 配置和 JavaConfig 驱动的配置。

The Hazelcast Event-Driven Inbound Channel Adapter listens to related cache events and sends event messages to the defined channel. It supports both XML and JavaConfig driven configurations.

XML Configuration :

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

Hazelcast 事件驱动的入站通道适配器需要以下属性：

The Hazelcast Event-Driven Inbound Channel Adapter requires the following attributes:

  • channel: Specifies the channel to which messages are sent;

  • cache: Specifies the distributed Object reference which is listened to. It is a mandatory attribute;

  • cache-events: Specifies cache events which are listened for. It is an optional attribute and its default value is ADDED. Its supported values are as follows :

  • Supported cache event types for IMap and MultiMap: ADDED, REMOVED, UPDATED, EVICTED, EVICT_ALL and CLEAR_ALL;

  • Supported cache event types for ReplicatedMap: ADDED, REMOVED, UPDATED, EVICTED;

  • Supported cache event types for IList, ISet and IQueue: ADDED, REMOVED. There are no cache event types for ITopic.

  • cache-listening-policy: Specifies the cache listening policy as SINGLE or ALL. It is an optional attribute and its default value is SINGLE. Each Hazelcast inbound channel adapter listening to the same cache object with the same cache-events attribute, can receive a single event message or all event messages. If it is ALL, all Hazelcast inbound channel adapters listening to the same cache object with the same cache-events attribute, will receive all event messages. If it is SINGLE, they will receive unique event messages.

一些配置示例：

Some configuration samples:

Distributed Map
<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>
Distributed MultiMap
<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>
Distributed List
<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>
Distributed Set
<int-hazelcast:inbound-channel-adapter channel="setChannel" cache="set" />

<bean id="set" factory-bean="instance" factory-method="getSet">
    <constructor-arg value="set"/>
</bean>
Distributed Queue
<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>
Distributed Topic
<int-hazelcast:inbound-channel-adapter channel="topicChannel" cache="topic" />

<bean id="topic" factory-bean="instance" factory-method="getTopic">
    <constructor-arg value="topic"/>
</bean>
Replicated Map
<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 Configuration Sample:

以下示例显示 DistributedMap 配置。相同配置可用于其他分布式数据结构（IMap、MultiMap、ReplicatedMap、IList、ISet、IQueue 和 ITopic）：

The following sample shows a DistributedMap configuration. The same configuration can be used for other distributed data structures(IMap, MultiMap, ReplicatedMap, IList, ISet, IQueue and ITopic):

@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 Continuous Query Inbound Channel Adapter

Hazelcast 连续查询允许对特定映射条目所做的修改进行侦听。Hazelcast 连续查询入站通道适配器是事件驱动的通道适配器，它根据已定义的谓语侦听相关的分布式映射事件。

Hazelcast Continuous Query enables listening to modifications performed on specific map entries. The Hazelcast Continuous Query Inbound Channel Adapter is an event-driven channel adapter which listens to the related distributed map events in the light of the defined predicate.

  • Java

  • XML

@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>

它支持六个属性，如下所示：

It supports six attributes as follows:

  • channel: Specifies the channel to which messages are sent;

  • cache: Specifies the distributed Map reference which is listened to. Mandatory;

  • cache-events: Specifies cache events which are listened for. Optional attribute with ADDED being its default value. Supported values are ADDED, REMOVED, UPDATED, EVICTED, EVICT_ALL and CLEAR_ALL;

  • predicate: Specifies a predicate to listen to the modifications performed on specific map entries. Mandatory;

  • include-value: Specifies including the value and oldValue in a continuous query result. Optional with true being the default;

  • cache-listening-policy: Specifies the cache listening policy as SINGLE or ALL. Optional with the default value being SINGLE. Each Hazelcast CQ inbound channel adapter listening to the same cache object with the same cache-events attribute, can receive a single event message or all event messages. If it is ALL, all Hazelcast CQ inbound channel adapters listening to the same cache object with the same cache-events attribute, will receive all event messages. If it is SINGLE, they will receive unique event messages.

Hazelcast Cluster Monitor Inbound Channel Adapter

Hazelcast 集群监视器支持侦听对集群所做的修改。Hazelcast 集群监视器入站通道适配器是事件驱动的通道适配器，它侦听相关的成员资格、分布式对象、迁移、生命周期和客户端事件：

A Hazelcast Cluster Monitor supports listening to modifications performed on the cluster. The Hazelcast Cluster Monitor Inbound Channel Adapter is an event-driven channel adapter and listens to related Membership, Distributed Object, Migration, Lifecycle and Client events:

  • Java

  • XML

@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>

它支持三个属性，如下所示：

It supports three attributes as follows :

  • channel: Specifies the channel to which messages are sent;

  • hazelcast-instance: Specifies the Hazelcast Instance reference to listen for cluster events. It is a mandatory attribute;

  • monitor-types: Specifies the monitor types which are listened for. It is an optional attribute with MEMBERSHIP being the default value. Supported values are MEMBERSHIP, DISTRIBUTED_OBJECT, MIGRATION, LIFECYCLE, CLIENT.

Hazelcast Distributed SQL Inbound Channel Adapter

Hazelcast 允许在分布式映射上运行分布式查询。Hazelcast 分布式 SQL 入站通道适配器是一种轮询入站通道适配器。它运行已定义的分布式 SQL 命令，并根据迭代类型返回结果。

Hazelcast allows running distributed queries on the distributed map. The Hazelcast Distributed SQL Inbound Channel Adapter is a polling inbound channel adapter. It runs the defined distributed-sql command and returns results depending on the iteration type.

  • Java

  • XML

@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>

它需要一个轮询器，并支持四个属性：

It requires a poller and supports four attributes:

  • channel: Specifies the channel to which messages are sent. It is a mandatory attribute;

  • cache: Specifies the distributed IMap reference which is queried. It is mandatory attribute;

  • iteration-type: Specifies result type. Distributed SQL can be run on EntrySet, KeySet, LocalKeySet or Values. It is an optional attribute with VALUE being the default. Supported values are ENTRY, `KEY, LOCAL_KEY and VALUE;

  • distributed-sql: Specifies the where clause of the sql statement. It is a mandatory attribute.

Hazelcast Outbound Channel Adapter

Hazelcast 出站通道适配器侦听其定义的通道，并将传入的消息写入到相关的分布式缓存。它期望其中一个：cache、cache-expression 或 HazelcastHeaders.CACHE_NAME 用于分布式对象定义。支持的分布式对象是：IMap、MultiMap、ReplicatedMap、IList、ISet、IQueue 和 ITopic。

The Hazelcast Outbound Channel Adapter listens to its defined channel and writes incoming messages to related distributed cache. It expects one of cache, cache-expression or HazelcastHeaders.CACHE_NAME for distributed object definition. Supported Distributed Objects are: IMap, MultiMap, ReplicatedMap, IList, ISet, IQueue and ITopic.

  • Java

  • XML

@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"/>

它需要以下属性：

It requires the following attributes :

  • channel: Specifies the channel to which messages are sent;

  • cache: Specifies the distributed object reference. Optional;

  • cache-expression: Specifies the distributed object via Spring Expression Language (SpEL). Optional;

  • key-expression: Specifies the key of a key-value pair via Spring Expression Language (SpEL). Optional and required for only for IMap, MultiMap and ReplicatedMap distributed data structures.

  • extract-payload: Specifies whether to send the whole message or just the payload. Optional attribute with true being the default. If it is true, just the payload will be written to the distributed object. Otherwise, the whole message will be written by converting both message headers and payload.

通过在头中设置分布式对象名称，可以经由同一信道将消息写入不同的分布式对象。如果未定义 cache 或 cache-expression 特性，必须在请求 Message 中设置 HazelcastHeaders.CACHE_NAME 头。

By setting distributed object name in the header, messages can be written to different distributed objects via same channel. If cache or cache-expression attributes are not defined, a HazelcastHeaders.CACHE_NAME header has to be set in a request Message.

Hazelcast Leader Election

如果需要领导人选举（例如，对于仅一个节点应该接收消息的高可用消息使用者而言），可以使用基于 Hazelcast 的 LeaderInitiator：

If leader election is needed (e.g. for highly available message consumer where only one node should receive messages) a Hazelcast-based LeaderInitiator can be used:

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

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

当一个节点被选为领导人时，它将向所有应用程序侦听器发送 OnGrantedEvent。

When a node is elected leader it will send an OnGrantedEvent to all application listeners.

Hazelcast Message Store

对于分布式消息状态管理（例如，对于持久性 QueueChannel 或跟踪 Aggregator 消息组），提供了 HazelcastMessageStore 实现：

For distributed messaging state management, for example for a persistent QueueChannel or tracking Aggregator message groups, the HazelcastMessageStore implementation is provided:

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

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

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

By default, the SPRING_INTEGRATION_MESSAGE_STORE IMap is used to store messages and groups as a key/value. Any custom IMap can be provided to the HazelcastMessageStore.

Hazelcast Metadata Store

实现一个 ListenableMetadataStore 可用，它使用后备 Hazelcast IMap。默认情况下，将使用名称为 SPRING_INTEGRATION_METADATA_STORE 的映射，该名称可以自定义。

An implementation of a ListenableMetadataStore is available using a backing Hazelcast IMap. The default map is created with a name SPRING_INTEGRATION_METADATA_STORE which can be customized.

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

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

HazelcastMetadataStore 实现 ListenableMetadataStore，允许你注册你自己的 MetadataStoreListener 类型的侦听器，以通过 addListener(MetadataStoreListener 回调) 侦听事件。

The HazelcastMetadataStore implements ListenableMetadataStore which allows you to register your own listeners of type MetadataStoreListener to listen for events via addListener(MetadataStoreListener callback).

Hazelcast Lock Registry

可以使用后备 Hazelcast 分布式 ILock 支持实现一个 LockRegistry：

An implementation of a LockRegistry is available using a backing Hazelcast distributed ILock support:

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

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

与共享 MessageGroupStore（例如，Aggregator 存储管理）配合使用时，HazelcastLockRegistry 可用于跨多个应用程序实例提供此功能，以便一次只允许一个实例处理组。

When used with a shared MessageGroupStore (e.g. Aggregator store management), the HazelcastLockRegistry can be used to provide this functionality across multiple application instances, such that only one instance can manipulate the group at a time.

对于所有分布式操作，必须在 HazelcastInstance 上启用 CP 子系统。

For all the distributed operations the CP Subsystem must be enabled on HazelcastInstance.

Message Channels with Hazelcast

Hazelcast IQueue 和 ITopic 分布式对象本质上是消息原语，可以与 Spring Integration 核心组件配合使用，而无需在此 Hazelcast 模块中实现额外的实现。

The Hazelcast IQueue and ITopic distributed objects are, essentially, messaging primitives and can be use with Spring Integration core components without extra implementations in this Hazelcast module.

QueueChannel 可由任何 java.util.Queue 提供，包括已提到的 Hazelcast 分布式 IQueue：

The QueueChannel can be supplied by any java.util.Queue, including the mentioned Hazelcast distributed IQueue:

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

将此配置放置在应用程序的 Hazelcast 集群中的多个节点上，将使 QueueChannel 得到分布，并且只有一个节点将能够从该 IQueue 轮询单个 Message。其工作方式类似于 PollableJmsChannel，PollableKafkaChannel 或 PollableAmqpChannel。

Placing this config on several nodes in Hazelcast cluster of the application, will make the QueueChannel as distributed and only one node will be able to poll a single Message from that IQueue. This works similar to PollableJmsChannel, PollableKafkaChannel or PollableAmqpChannel.

如果生产者端不是 Spring 集成应用程序，则无法配置 QueueChannel，因此使用普通 Hazelcast IQueue API 产生数据。在这种情况下，QueueChannel 方法在消费者端是错误的：必须改用 Inbound Channel Adapter 解决方案：

If the producer side is not a Spring Integration application, there is no way to configure a QueueChannel, and therefore the plain Hazelcast IQueue API is used to produce the data. In this case, the QueueChannel approach is wrong on the consumer side: an Inbound Channel Adapter solution must be used instead:

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

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

Hazelcast 中的 ITopic 抽象与 JMS 中的 Topic 具有相似的语义：所有订户都收到发布的消息。使用一对简单的 MessageChannel bean，此机制作为开箱即用的功能得到支持：

The ITopic abstraction in Hazelcast has similar semantics to a Topic in JMS: all subscribers receive published messages. With a pair of simple MessageChannel beans this mechanism is supported as an out-of-the-box feature:

@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 是一个函数式接口，因此可以为 handleMessage 方法提供一个简单的 lambda。当消息发送到 publishToHazelcastTopicChannel 时，它只是发布到 Hazelcast ITopic。com.hazelcast.topic.MessageListener 也是一个函数式接口，因此可以为 ITopic#addMessageListener 提供一个 lambda。因此，fromHazelcastTopicChannel 订阅者将使用所有发送到所述 ITopic 的消息。

The FixedSubscriberChannel is an optimized variant of DirectChannel, which requires a MessageHandler on initialization. Since the MessageHandler is a functional interface a simple lambda for the handleMessage method can be provided. When a message is sent to the publishToHazelcastTopicChannel it is just published onto the Hazelcast ITopic. The com.hazelcast.topic.MessageListener is a functional interface, too, hence a lambda to the ITopic#addMessageListener can be provided. So, a subscriber to the fromHazelcastTopicChannel will consume all messages sent to the mentioned ITopic.

可以将 ExecutorChannel 与 IExecutorService 一起提供。例如，通过相应的配置，可以实现集群范围内单例：

An ExecutorChannel can be supplied with an IExecutorService. For example, with respective configuration a cluster-wide singleton can be achieved:

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