当前位置：首页 > article >正文

在Spring Boot项目中使用Zookeeper和Curator实现高效、可靠的分布式锁

article 2025/2/28 23:39:11

要在 Spring Boot 项目中使用 Zookeeper 和 Curator 实现高效、可靠的分布式锁，可以参考以下步骤和优化建议：

1. 引入依赖

在 pom.xml 中添加 Curator 和 Zookeeper 相关依赖：

<dependencies>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter</artifactId>
    </dependency>
    <dependency>
        <groupId>org.apache.curator</groupId>
        <artifactId>curator-framework</artifactId>
        <version>5.1.0</version>
    </dependency>
    <dependency>
        <groupId>org.apache.curator</groupId>
        <artifactId>curator-recipes</artifactId>
        <version>5.1.0</version>
    </dependency>
    <dependency>
        <groupId>org.apache.zookeeper</groupId>
        <artifactId>zookeeper</artifactId>
        <version>3.8.0</version>
    </dependency>
</dependencies>

2. 配置 Zookeeper 连接

在 application.yml 中配置 Zookeeper 连接字符串：

zookeeper:
  connect-string: localhost:2181

3. 配置 Curator 客户端

创建一个配置类来初始化 Curator 客户端：

import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.retry.RetryNTimes;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class ZookeeperConfig {

    @Value("${zookeeper.connect-string}")
    private String connectString;

    @Bean(initMethod = "start", destroyMethod = "close")
    public CuratorFramework curatorFramework() {
        return CuratorFrameworkFactory.builder()
             .connectString(connectString)
             .sessionTimeoutMs(5000)
             .retryPolicy(new RetryNTimes(3, 5000))
             .build();
    }
}

4. 实现分布式锁服务

创建一个服务类来封装分布式锁的获取和释放逻辑，并进行一些优化：

import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.recipes.locks.InterProcessMutex;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.util.concurrent.TimeUnit;

@Service
public class DistributedLockService {

    private static final Logger logger = LoggerFactory.getLogger(DistributedLockService.class);

    @Autowired
    private CuratorFramework curatorFramework;

    // 重试获取锁的次数
    private static final int MAX_RETRIES = 3;
    // 每次重试的间隔时间
    private static final int RETRY_INTERVAL_MS = 1000;

    // 获取分布式锁
    public boolean acquireLock(String lockPath, int timeout, TimeUnit timeUnit) {
        InterProcessMutex lock = new InterProcessMutex(curatorFramework, lockPath);
        for (int i = 0; i < MAX_RETRIES; i++) {
            try {
                if (lock.acquire(timeout, timeUnit)) {
                    logger.info("成功获取分布式锁，锁路径: {}", lockPath);
                    return true;
                }
            } catch (Exception e) {
                logger.error("获取分布式锁失败，尝试重试，锁路径: {}", lockPath, e);
            }
            try {
                TimeUnit.MILLISECONDS.sleep(RETRY_INTERVAL_MS);
            } catch (InterruptedException e) {
                logger.error("重试间隔被中断", e);
                Thread.currentThread().interrupt();
            }
        }
        logger.warn("经过多次重试仍无法获取分布式锁，锁路径: {}", lockPath);
        return false;
    }

    // 释放分布式锁
    public void releaseLock(String lockPath) {
        InterProcessMutex lock = new InterProcessMutex(curatorFramework, lockPath);
        try {
            lock.release();
            logger.info("成功释放分布式锁，锁路径: {}", lockPath);
        } catch (Exception e) {
            logger.error("释放分布式锁失败，锁路径: {}", lockPath, e);
        }
    }
}

5. 使用分布式锁

在需要使用分布式锁的地方注入 DistributedLockService 并调用相应方法：

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;

@RestController
public class DistributedLockUsageController {

    @Autowired
    private DistributedLockService distributedLockService;

    @GetMapping("/protected-operation")
    public String protectedOperation(@RequestParam String lockPath) {
        int timeout = 10;
        TimeUnit timeUnit = TimeUnit.SECONDS;
        if (distributedLockService.acquireLock(lockPath, timeout, timeUnit)) {
            try {
                // 这里放置需要同步执行的业务逻辑
                return "成功获取锁并执行了受保护的操作";
            } finally {
                distributedLockService.releaseLock(lockPath);
            }
        } else {
            return "无法获取锁，操作被拒绝";
        }
    }
}