SpringBoot整合Zookeeper做分布式锁

环境准备

zookeeper准备

首先你需要一个zookeeper服务器，或者是一个zookeeper集群。我已经准备好了一个zookeeper集群，如图：

当然一个单节点的zookeeper也可以搭建分布式锁。如果你还没有zookeeper，那么你可以参考我写的搭建zookeeper集群的文章：https://blog.csdn.net/m0_51510236/article/details/132834141

SpringBoot项目准备

这个步骤比较简单，我的代码已经提交至公共代码仓库，代码仓库地址为：https://gitcode.net/m0_51510236/zookeeper-distribution-lock

代码编写

pom.xml依赖文件

首先我们需要添加一下SpringCloudZookeeper的依赖，因为是只使用到了SpringCloud的zookeeper模块，所以没必要引入整个SpringCloud，如果你已经引入了整个SpringCloud，那么就没必要引入该模块了，如图：

其次我们要在dependency里面引入zookeeper的依赖：

<!-- Zookeeper 客户端 -->
<dependency>
    <groupId>org.springframework.cloud</groupId>
    <artifactId>spring-cloud-starter-zookeeper</artifactId>
</dependency>

<!-- Zookeeper 分布式锁 -->
<dependency>
    <groupId>org.apache.curator</groupId>
    <artifactId>curator-recipes</artifactId>
</dependency>
1
2
3
4
5
6
7
8
9
10
11

然后我们还需要引入SpringBoot测试模块的依赖：

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-test</artifactId>
    <scope>test</scope>
</dependency>
1
2
3
4
5

整体pom.xml文件：

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
    xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>
    <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>2.6.15</version>
        <relativePath/> <!-- lookup parent from repository -->
    </parent>
    <groupId>city.yueyang</groupId>
    <artifactId>zookeeper-distribution-lock</artifactId>
    <version>0.0.1-SNAPSHOT</version>
    <name>zookeeper-distribution-lock</name>
    <description>zookeeper分布式锁</description>
    <properties>
        <java.version>11</java.version>
    </properties>
    <dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter</artifactId>
        </dependency>

        <!-- Zookeeper 客户端 -->
        <dependency>
            <groupId>org.springframework.cloud</groupId>
            <artifactId>spring-cloud-starter-zookeeper</artifactId>
        </dependency>

        <!-- Zookeeper 分布式锁 -->
        <dependency>
            <groupId>org.apache.curator</groupId>
            <artifactId>curator-recipes</artifactId>
        </dependency>

        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
        </dependency>
    </dependencies>

    <dependencyManagement>
        <dependencies>
            <dependency>
                <groupId>org.springframework.cloud</groupId>
                <artifactId>spring-cloud-zookeeper-dependencies</artifactId>
                <version>3.1.4</version>
                <scope>import</scope>
                <type>pom</type>
            </dependency>
        </dependencies>
    </dependencyManagement>

    <build>
        <plugins>
            <plugin>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-maven-plugin</artifactId>
            </plugin>
        </plugins>
    </build>
</project>
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64

application.yaml文件

application.yaml是SpringCloud的配置文件，我们只需要配置zookeeper的地址即可，文件内容为：

spring:
  cloud:
    zookeeper:
      # 可以只写一个，如果是有多个节点的zookeeper集群，那么多个节点用英文逗号隔开
      connect-string: 192.168.1.181:2181,192.168.1.182:2181,192.168.1.183:2181
1
2
3
4
5

Runnable类

我们需要建立一个线程池来模拟多线程整合zookeeper做分布式锁，那么我们需要一个Runnable类来定义多线程的任务，java代码为：

package city.yueyang.lock.thread;

import org.apache.curator.framework.recipes.locks.InterProcessMutex;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.CountDownLatch;

/**
 * <p></p>
 *
 * @author XiaoHH
 * @version 1.0.0
 * @date 2023-09-13 星期三 20:44:59
 * @file DistributionLockRunnable.java
 */
public class DistributionLockRunnable implements Runnable {

    private static final Logger log = LoggerFactory.getLogger(DistributionLockRunnable.class);

    /**
     * 分布式锁对象
     */
    private final InterProcessMutex lock;

    /**
     * CountDownLatch锁，避免多线程没有执行完就退出程序
     */
    private final CountDownLatch countDownLatch;

    public DistributionLockRunnable(InterProcessMutex lock, CountDownLatch countDownLatch) {
        this.lock = lock;
        this.countDownLatch = countDownLatch;
    }

    @Override
    public void run() {
        String threadName = Thread.currentThread().getName();
        try {
            log.info("线程名：{}，尝试获取锁", threadName);
            // 获取锁
            lock.acquire();
            log.info("线程名：{}，获取锁成功，正在处理数据", threadName);
            Thread.sleep(1000); // 模拟处理数据睡眠一秒钟
            log.info("线程名：{}，数据处理成功", threadName);
        } catch (Throwable e) {
            log.error("线程名：{}，发生了错误", threadName, e);
        } finally {
            try {
                // 退出锁
                lock.release();
                log.info("线程名：{}，锁已释放", threadName);
                // 锁-1
                countDownLatch.countDown();
            } catch (Exception e) {
                log.error("线程名：{}，释放锁的时候发生了错误", threadName, e);
            }
        }
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60

注意下面两行代码：

lock.acquire(); // 获取锁
lock.release(); // 释放锁
1
2

代码里面日志打印已经很详细，这里就不再过多解释

测试类编写

我们编写一个测试用例来测试这个分布式锁：

package city.yueyang.lock;

import city.yueyang.lock.thread.DistributionLockRunnable;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.recipes.locks.InterProcessMutex;
import org.junit.jupiter.api.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;

import java.util.concurrent.*;

@SpringBootTest
public class ZookeeperDistributionLockApplicationTests {

    /**
     * 日志记录对象
     */
    private static final Logger log = LoggerFactory.getLogger(ZookeeperDistributionLockApplicationTests.class);

    /**
     * zookeeper的客户端
     */
    @Autowired
    private CuratorFramework curatorFramework;

    /**
     * zookeeper锁的路径，这也会被认为是zookeeper锁的标识
     */
    private static final String LOCK_PATH = "/lock/test-zookeeper-lock";

    @Test
    public void testDistributionLock() {
        // 处理十个任务
        final int taskAmount = 10;
        // 使用 CountDownLatch 锁避免多线程没有执行完程序就停止
        CountDownLatch countDownLatch = new CountDownLatch(taskAmount);
        // 创建zookeeper的锁对象，如果第二个参数也就是path路径是一样的，那么就会被认为是同一把锁
        InterProcessMutex lock = new InterProcessMutex(this.curatorFramework, LOCK_PATH);
        // 创建线程池
        ThreadPoolExecutor executor = new ThreadPoolExecutor(taskAmount, taskAmount, 100, TimeUnit.SECONDS, new ArrayBlockingQueue<>(taskAmount));
        for (int i = 0; i < taskAmount; i++) {
            // 创建一个任务
            Runnable task = new DistributionLockRunnable(lock, countDownLatch);
            // 使用线程池去执行它
            executor.execute(task);
        }
        try {
            // 等待所有的countDown锁执行完毕本线程结束
            countDownLatch.await();
        } catch (InterruptedException e) {
            log.error("countdown锁被终止了");
        } finally {
            // 终止线程池
            executor.shutdown();
        }
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59

创建锁对象的代码主要是下面这行：

InterProcessMutex lock = new InterProcessMutex(this.curatorFramework, "锁路径的字符串");
1

重点是第二个参数，是锁路径。无论new出多少个锁对象，只要锁路径的字符串是一样的，就会被zookeeper认为是同一把锁

测试代码

我们直接运行这个测试用例，可以发现都是有序的获取锁，并没有并发执行的问题，这也证明了分布式锁搭建成功：

代码已提交到公共代码仓库。代码仓库地址：https://gitcode.net/m0_51510236/zookeeper-distribution-lock