Spring Integration Zookeeper 支持教程

引言：分布式协调利器 Zookeeper

NOTE

Zookeeper 是分布式系统的协调服务，提供配置维护、命名服务、分布式同步等功能。Spring Integration 4.2+ 提供了对 Zookeeper 的深度集成，简化分布式应用开发。

核心功能概览

依赖配置

在 build.gradle.kts 中添加：

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

TIP

使用最新版本可获取更好的性能和稳定性支持

---

一、Zookeeper 元数据存储

核心概念

元数据存储 (MetadataStore) 用于持久化关键数据，如文件处理状态、作业进度等。Zookeeper 的树形结构非常适合此类场景。

Kotlin 配置示例

@Configuration
class ZookeeperConfig {

    // 创建 CuratorFramework 客户端
    @Bean
    fun zkClient(): CuratorFramework {
        return CuratorFrameworkFactory.newClient(
            "localhost:2181", 
            RetryNTimes(3, 1000)
        ).apply { start() }
    }

    // 配置元数据存储
    @Bean
    fun metadataStore(client: CuratorFramework): MetadataStore {
        return ZookeeperMetadataStore(client).apply {
            // 设置根路径 (可选)
            setRoot("/myApp/metadata")
        }
    }
}

// [!code highlight] 关键配置点：连接字符串和重试策略

使用场景

@Service
class FileProcessorService(
    private val metadataStore: MetadataStore
) {
    fun processFile(file: File) {
        val lastModified = metadataStore.get("file_${file.name}")
        if (lastModified != file.lastModified().toString()) {
            // 处理新文件
            metadataStore.put("file_${file.name}", file.lastModified().toString())
        }
    }
}

IMPORTANT

Zookeeper 对小数据(<1MB)支持最佳，大文件应存储在专用存储系统中

---

二、Zookeeper 分布式锁

分布式锁原理

在集群环境中协调资源访问，确保同一时间只有一个节点能执行关键操作。

Kotlin 配置

@Configuration
class LockConfig {

    @Bean
    fun lockRegistry(client: CuratorFramework): LockRegistry {
        return ZookeeperLockRegistry(client).apply {
            // 设置缓存容量 (5.5.6+)
            cacheCapacity = 1000
        }
    }
}

使用示例：集群任务调度

@Service
class ClusterTaskScheduler(
    private val lockRegistry: LockRegistry
) {
    fun executeScheduledTask() {
        val lock = lockRegistry.obtain("dailyReportLock")
        if (lock.tryLock()) {
            try {
                // 确保集群中只有一个节点执行
                generateDailyReport()
            } finally {
                lock.unlock()
            }
        }
    }
}

死锁风险

总是使用 try-finally 确保锁释放，避免死锁

锁路径策略

自定义锁在 Zookeeper 中的存储路径：

class CustomPathStrategy : KeyToPathStrategy {
    override fun pathFor(key: String): String {
        return "/locks/${key.hashCode()}"
    }

    override fun bounded() = true // 标记为有界，无需清理
}

---

三、领导选举与事件处理

领导选举机制

在集群中自动选举领导者节点，非领导者节点处于待命状态。

Kotlin 配置

@Configuration
class LeadershipConfig {

    @Bean
    fun leaderInitiator(client: CuratorFramework): LeaderInitiatorFactoryBean {
        return LeaderInitiatorFactoryBean().apply {
            setClient(client)
            setPath("/myApp/leader")
            setRole("primary") // 定义角色名称
        }
    }
}

领导事件处理

@Component
class LeadershipEventHandler {

    @EventListener
    fun handleGranted(event: OnGrantedEvent) {
        if ("primary" == event.role) {
            // 成为领导者时启动服务
            startPrimaryServices()
        }
    }

    @EventListener
    fun handleRevoked(event: OnRevokedEvent) {
        if ("primary" == event.role) {
            // 失去领导权时停止服务
            stopPrimaryServices()
        }
    }
}

领导选举流程

---

四、常见问题解决方案

问题1：Zookeeper 连接不稳定

解决方案：

@Bean
fun zkClient(): CuratorFramework {
    return CuratorFrameworkFactory.builder()
        .connectString("zk1:2181,zk2:2181,zk3:2181") // 多节点配置
        .retryPolicy(ExponentialBackoffRetry(1000, 5))
        .build()
        .apply { start() }
}

问题2：领导切换频繁

优化方案：

leaderInitiator.setLeaderSelector(LeaderSelector( 
    client, 
    "/app/leader", 
    LeaderSelectorListener { context ->
        // 获取领导权后保持，直到主动释放
        while (context.hasLeadership()) {
            Thread.sleep(5000) // 心跳检测
        }
    }
))

CAUTION

避免在领导者角色中执行阻塞操作，可能导致会话超时

问题3：锁性能瓶颈

优化策略：

• 使用细粒度锁（按资源ID而不是全局锁）
• 设置合理的锁超时时间
• 定期清理无用锁：

  lockRegistry.expireUnusedOlderThan(TimeUnit.MINUTES.toMillis(30))

---

最佳实践总结

1. 连接管理：使用连接池和重试策略增强鲁棒性
2. 数据大小：Zookeeper 节点数据保持 <1KB
3. 监听优化：使用 PathChildrenCache 替代直接监听
4. 安全配置：启用 SASL 认证

   CuratorFrameworkFactory.builder()
        .authorization("digest", "user:password".toByteArray())

生产环境建议

部署 Zookeeper 集群（至少3节点），配置自动故障转移和定期快照清理

[!SUCCESS] 通过 Spring Integration 的 Zookeeper 支持，开发者可以轻松构建高可用、强一致的分布式系统，而无需深入 Zookeeper 复杂 API 细节。