Spring Integration Hazelcast 集成指南

概述

Spring Integration 提供了一套完整的组件，用于与 Hazelcast 内存数据网格进行交互。Hazelcast 是一个分布式内存计算平台，能够处理大规模数据并提供低延迟访问。

核心价值

通过 Spring Integration 的 Hazelcast 支持，开发者可以：

• 轻松监听分布式数据结构的变化
• 执行分布式 SQL 查询
• 实现集群监控和领导选举
• 构建弹性的分布式消息系统

依赖配置

在项目中添加 Hazelcast 集成依赖：

Gradle

dependencies {
    implementation("org.springframework.integration:spring-integration-hazelcast:6.5.1")
}

Maven

<dependency>
    <groupId>org.springframework.integration</groupId>
    <artifactId>spring-integration-hazelcast</artifactId>
    <version>6.5.1</version>
</dependency>

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

数据结构支持

Hazelcast 提供多种分布式数据结构：

Kotlin 配置示例

@Bean
fun distributedMapChannel(): PollableChannel = QueueChannel()

@Bean
fun distributedMap(): IMap<Int, String> = hazelcastInstance().getMap("Distributed_Map")

@Bean
fun hazelcastInstance(): HazelcastInstance = Hazelcast.newHazelcastInstance()

@Bean
fun hazelcastEventDrivenMessageProducer(): HazelcastEventDrivenMessageProducer {
    return HazelcastEventDrivenMessageProducer(distributedMap()).apply {
        outputChannel = distributedMapChannel()
        setCacheEventTypes("ADDED,REMOVED,UPDATED,CLEAR_ALL")
        cacheListeningPolicy = CacheListeningPolicyType.SINGLE
    }
}

配置选项详解

参数	说明	默认值
cache	监听的分布式对象	必填
cache-events	监听的事件类型	ADDED
cache-listening-policy	监听策略 (SINGLE/ALL)	SINGLE

注意事项

• ITopic 没有特定的事件类型
• ReplicatedMap 不支持 EVICT_ALL 和 CLEAR_ALL 事件

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

工作原理

Kotlin 配置示例

@Bean
fun cqDistributedMapChannel(): PollableChannel = QueueChannel()

@Bean
fun cqDistributedMap(): IMap<Int, Person> = hazelcastInstance().getMap("CQ_Map")

@Bean
fun hazelcastContinuousQueryMessageProducer(): HazelcastContinuousQueryMessageProducer {
    return HazelcastContinuousQueryMessageProducer(
        cqDistributedMap(), 
        "surname='TestSurname'"
    ).apply {
        outputChannel = cqDistributedMapChannel()
        setCacheEventTypes("UPDATED")
        includeValue = false
    }
}

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

监控事件类型

事件类型	说明
MEMBERSHIP	集群成员变更
DISTRIBUTED_OBJECT	分布式对象变更
MIGRATION	数据迁移事件
LIFECYCLE	实例生命周期
CLIENT	客户端连接事件

Kotlin 配置示例

@Bean
fun eventChannel(): PollableChannel = QueueChannel()

@Bean
fun hazelcastClusterMonitorMessageProducer(): HazelcastClusterMonitorMessageProducer {
    return HazelcastClusterMonitorMessageProducer(hazelcastInstance()).apply {
        outputChannel = eventChannel()
        monitorEventTypes = "DISTRIBUTED_OBJECT"
    }
}

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

迭代类型对比

类型	返回结果	适用场景
ENTRY	键值对实体	需要完整数据
KEY	键集合	仅需标识符
LOCAL_KEY	本地键集合	优化本地查询
VALUE	值集合	仅需数据内容

Kotlin 配置示例

@Bean
fun dsDistributedMapChannel(): PollableChannel = QueueChannel()

@Bean
@InboundChannelAdapter(
    value = "dsDistributedMapChannel", 
    poller = [Poller(maxMessagesPerPoll = "1")]
)
fun hazelcastDistributedSQLMessageSource(): HazelcastDistributedSQLMessageSource {
    return HazelcastDistributedSQLMessageSource(
        dsDistributedMap(),
        "name='TestName' AND surname='TestSurname'"
    ).apply {
        iterationType = DistributedSQLIterationType.ENTRY
    }
}

Hazelcast 出站通道适配器

数据写入流程

Kotlin 配置示例

@Bean
fun distributedMapChannel(): MessageChannel = DirectChannel()

@ServiceActivator(inputChannel = "distributedMapChannel")
fun hazelcastCacheWritingMessageHandler(): HazelcastCacheWritingMessageHandler {
    return HazelcastCacheWritingMessageHandler().apply {
        distributedObject = distributedMap()
        keyExpression = SpelExpressionParser().parseExpression("payload.id")
        extractPayload = true
    }
}

Hazelcast 消息存储

消息存储实现

@Bean
fun messageStore(): MessageGroupStore {
    return HazelcastMessageStore(hazelcastInstance())
}

@Bean
fun metadataStore(): MetadataStore {
    return HazelcastMetadataStore(hazelcastInstance())
}

最佳实践

• 使用 HazelcastMessageStore 存储消息组状态
• 使用 HazelcastMetadataStore 存储元数据
• 自定义 IMap 名称以隔离不同应用

Hazelcast 消息通道

队列通道实现

@Bean
fun hazelcastQueueChannel(): PollableChannel {
    return QueueChannel(hazelcastInstance().getQueue("springIntegrationQueue"))
}

主题通道实现

@Bean
fun springIntegrationTopic(): ITopic<Message<*>> {
    val topic = hazelcastInstance().getTopic<Message<*>>("springIntegrationTopic")
    topic.addMessageListener { 
        fromHazelcastTopicChannel().send(it.messageObject) 
    }
    return topic
}

@Bean
fun publishToHazelcastTopicChannel(): MessageChannel {
    return FixedSubscriberChannel { message -> 
        springIntegrationTopic().publish(message) 
    }
}

弃用组件说明

弃用通知

以下组件已在 Hazelcast 5.5+ 中弃用：

• LeaderInitiator
• HazelcastLockRegistry

替代方案：

// 使用 Hazelcast 原生的 CP 子系统
val lock = hazelcastInstance.cpSubsystem.getLock("myLock")
lock.lock()
try {
    // 关键代码段
} finally {
    lock.unlock()
}

常见问题解决

问题1：事件监听器不触发

可能原因：

• 未正确配置缓存事件类型
• Hazelcast 实例未加入集群

解决方案：

// 确保配置了正确的事件类型
producer.setCacheEventTypes("ADDED,UPDATED,REMOVED")

// 验证 Hazelcast 实例状态
if (hazelcastInstance().cluster.members.isEmpty()) {
    logger.error("Hazelcast 实例未加入集群")
}

问题2：分布式查询性能低下

优化建议：

// 使用本地键集迭代减少网络传输
messageSource.iterationType = DistributedSQLIterationType.LOCAL_KEY

// 添加索引提升查询性能
val mapConfig = hazelcastInstance().config.getMapConfig("myMap")
mapConfig.addIndexConfig(IndexConfig(IndexType.HASH, "name"))

总结

通过 Spring Integration 的 Hazelcast 支持，开发者可以轻松构建分布式、高可用的系统。关键要点：

1. 入站适配器：监听数据结构变化和集群事件
2. 出站适配器：写入数据到分布式存储
3. 消息存储：实现分布式状态管理
4. 通道集成：构建弹性消息传递系统

生产环境建议

• 使用 CacheListeningPolicyType.ALL 确保事件不丢失
• 为频繁查询的字段添加索引
• 监控 Hazelcast 集群健康状态
• 使用 ReplicatedMap 替代 IMap 实现更高可用性

通过本教程，您应该能够掌握 Spring Integration 与 Hazelcast 集成的核心概念和实现方式。在实际应用中，根据具体需求选择合适的组件和配置策略。