Spring Boot 3 集成 ZooKeeper：从环境搭建到实战应用-拓冰建站

1. ZooKeeper 简介与 Spring Boot 集成价值

ZooKeeper 是一个分布式的、开源的分布式应用程序协调服务，它主要用于维护配置信息、命名、提供分布式同步和提供组服务。许多分布式应用（如 Hadoop、Kafka、HBase 等）都依赖 ZooKeeper 来实现核心的协调功能。

Spring Boot 3 通过 Spring Integration Zookeeper 模块或 Apache Curator 客户端，可以方便地与 ZooKeeper 集成，实现配置中心、分布式锁、服务注册与发现等常见场景。

2. ZooKeeper 环境搭建与配置

1. 修改配置文件

# 1、拷贝一份样本配置文件 cp zookeeper-3.8.3/conf/zoo_sample.cfg zookeeper-3.8.3/conf/zoo.cfg 2、修改数据文件地址，注意这里用本地路径 dataDir=/local-path/zookeeper-3.8.3/data 3、添加一个配置，处理启动日志的提示：ZooKeeper audit is disabled. audit.enable=true

2. 服务启动

# 1、启动服务端 zookeeper-3.8.3/bin/zkServer.sh start 2、停止服务端 zookeeper-3.8.3/bin/zkServer.sh stop 3、启动客户端 zookeeper-3.8.3/bin/zkCli.sh

3. 客户端测几个增删查的命令

[zk: localhost:2181(CONNECTED) 0] create /cicada smile1 Created /cicada [zk: localhost:2181(CONNECTED) 1] get /cicada smile1 [zk: localhost:2181(CONNECTED) 2] ls / [cicada, zookeeper] [zk: localhost:2181(CONNECTED) 3] delete /cicada

3. Spring Boot 3 工程集成 ZooKeeper

1. 工程结构

一个典型的 Spring Boot 3 项目结构如下：

src/main/java └── com └── example └── zkdemo ├── ZookeeperDemoApplication.java ├── config │ └── ZookeeperConfig.java └── service └── ZookeeperService.java src/main/resources ├── application.yml └── logback-spring.xml

2. 依赖管理

在pom.xml中添加以下依赖：

<!-- Zookeeper 客户端 --> <dependency> <groupId>org.apache.zookeeper</groupId> <artifactId>zookeeper</artifactId> <version>3.8.3</version> </dependency> <!-- Curator 客户端（Spring Integration Zookeeper 内部已包含） --> <dependency> <groupId>org.springframework.integration</groupId> <artifactId>spring-integration-zookeeper</artifactId> <version>6.1.0</version> </dependency> <!-- Spring Boot Starter Test --> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-test</artifactId> <scope>test</scope> </dependency>

注意：Curator 是一组让 ZooKeeper 使用更简单的 Java 库。这里使用spring-integration-zookeeper依赖，它内部已经封装了 Curator 客户端，并提供了与 Spring 生态更好的集成方式。

3. 配置文件

application.yml：

zookeeper: # 服务器地址 connectString: 127.0.0.1:2181 # 会话超时时间（毫秒） sessionTimeoutMs: 3000 # 连接超时时间（毫秒） connectionTimeoutMs: 60000 # 最大重试次数 maxRetries: 3 # 初始休眠时间（毫秒） baseSleepTimeMs: 1000

配置类 ZookeeperConfig.java：

import org.apache.curator.RetryPolicy; import org.apache.curator.framework.CuratorFramework; import org.apache.curator.framework.CuratorFrameworkFactory; import org.apache.curator.retry.ExponentialBackoffRetry; import org.springframework.beans.factory.annotation.Value; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import javax.annotation.PostConstruct; @Configuration public class ZookeeperConfig { @Value("${zookeeper.connectString}") private String connectString; @Value("${zookeeper.baseSleepTimeMs}") private int baseSleepTimeMs; @Value("${zookeeper.maxRetries}") private int maxRetries; @Value("${zookeeper.connectionTimeoutMs}") private int connectionTimeoutMs; @Value("${zookeeper.sessionTimeoutMs}") private int sessionTimeoutMs; private CuratorFramework client; /** 初始化 Curator 客户端 */ @PostConstruct public void init() { // 重试策略：基础休眠时间 1000ms，最大重试 3 次 RetryPolicy retryPolicy = new ExponentialBackoffRetry(baseSleepTimeMs, maxRetries); // 创建 CuratorFramework 实例 client = CuratorFrameworkFactory.builder() .connectString(connectString) .connectionTimeoutMs(connectionTimeoutMs) .sessionTimeoutMs(sessionTimeoutMs) .retryPolicy(retryPolicy) .build(); // 开启连接 client.start(); } @Bean public CuratorFramework curatorFramework() { return client; } }

4. ZooKeeper 核心 API 实战示例

下面通过单元测试演示常见的 ZooKeeper API 操作，包括节点创建、数据读写、节点删除以及分布式读写锁的使用。

import org.apache.curator.framework.CuratorFramework; import org.apache.curator.framework.recipes.locks.InterProcessReadWriteLock; import org.apache.zookeeper.CreateMode; import org.apache.zookeeper.data.Stat; import org.junit.jupiter.api.Test; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.boot.test.context.SpringBootTest; import java.nio.charset.StandardCharsets; @SpringBootTest public class ZookeeperApiTest { @Autowired private CuratorFramework client; @Test public void testCreate() throws Exception { // 创建一个持久化节点，断开连接时不会自动删除 client.create() .creatingParentsIfNeeded() .withMode(CreateMode.PERSISTENT) .forPath("/path1"); System.out.println("持久化节点 /path1 创建成功"); } @Test public void testExists() throws Exception { // 判断节点是否存在 Stat stat1 = client.checkExists().forPath("/path1"); System.out.println("/path1 是否存在: " + (stat1 != null)); Stat stat2 = client.checkExists().forPath("/path2"); System.out.println("/path2 是否存在: " + (stat2 != null)); } @Test public void testSetData() throws Exception { // 设置节点数据 client.setData() .forPath("/path1", "data1".getBytes(StandardCharsets.UTF_8)); System.out.println("节点 /path1 数据设置成功"); } @Test public void testCreateAndSet() throws Exception { // 创建一个持久化节点并设置节点数据 client.create() .creatingParentsIfNeeded() .withMode(CreateMode.PERSISTENT) .forPath("/path3", "data3".getBytes(StandardCharsets.UTF_8)); System.out.println("持久化节点 /path3 创建并设置数据成功"); } @Test public void testGetData() throws Exception { // 查询节点数据 byte[] data = client.getData().forPath("/path3"); String dataStr = new String(data, StandardCharsets.UTF_8); System.out.println("节点 /path3 的数据: " + dataStr); } @Test public void testDelete() throws Exception { // 删除节点（同时删除子节点） client.delete() .guaranteed() .deletingChildrenIfNeeded() .forPath("/path3"); System.out.println("节点 /path3 删除成功"); } @Test public void testReadLock() throws Exception { // 读写锁 - 读锁 InterProcessReadWriteLock lock = new InterProcessReadWriteLock(client, "/lock-read"); lock.readLock().acquire(); System.out.println("获取 ReadLock 成功"); // 模拟读操作 Thread.sleep(1000); lock.readLock().release(); System.out.println("释放 ReadLock"); } @Test public void testWriteLock() throws Exception { // 读写锁 - 写锁 InterProcessReadWriteLock lock = new InterProcessReadWriteLock(client, "/lock-write"); lock.writeLock().acquire(); System.out.println("获取 WriteLock 成功"); // 模拟写操作 Thread.sleep(1000); lock.writeLock().release(); System.out.println("释放 WriteLock"); } }