基于 ZooKeeper 原生 API 实现分布式锁

发表于 更新于 阅读次数： 评论数： 本文字数： 3.1k 阅读时长 ≈ 3 分钟

大纲

前言

本文将基于 ZooKeeper 原生 API 实现分布式锁，包括公平锁和非公平锁两种类型。值得注意的是，文中给出的代码缺乏健壮性和可用性（尤其是对网络抖动、可重入、会话过期的处理），仅供参考学习，不适用于生产环境。在生产环境中，强烈建议通过 Apache Curator 提供的 API 来实现基于 ZooKeeper 的分布式锁。

Apache Curator 介绍

Apache Curator 是 ZooKeeper 的客户端工具包，封装了很多常见的分布式功能（Recipes），比如：分布式锁、分布式队列、分布式计数器、领导者选举（Leader Election）、单向栅栏（Barrier）、双向栅栏（DoubleBarrier）、组成员管理（Group Membership）等。

核心概念

ZK 实现分布式锁的几种方式

ZooKeeper 实现分布式锁有两种常见的方式：

  • (1) 基于临时节点实现分布式锁（非公平锁）

    • 客户端在指定的父节点下创建一个临时节点，如果成功创建，则获得锁；如果抛出异常，说明锁已经被占用，那么就监听已创建的节点，并等待接收监听事件通知。
    • 当锁的持有者删除自己创建的节点，从而释放锁。ZooKeeper 会通知在该节点上监听的客户端，该客户端会重新尝试创建一个同名的临时节点，若创建成功，则获得锁。
    • 该方式可以理解为基于异常的分布式锁实现，缺点是存在惊群效应的问题，也就是多个客户端可能同时尝试获取锁，导致频繁地创建节点和设置监视器。

  • (2) 基于临时顺序节点实现分布式锁（公平锁）

    • 客户端在指定的父节点下创建一个有序的临时节点，然后获取该父节点下所有的子节点，并检查自己创建的节点是否是最小的序号节点。
    • 如果是，则该客户端获得锁。如果不是，则客户端给序号最接近且小于自己的那个节点设置监视器（Watcher），然后进入等待状态。
    • 持有锁的客户端执行完临界区代码（即操作共享数据的代码）后，删除自己创建的节点，从而释放锁。
    • 当持有锁的客户端删除节点后，ZooKeeper 会通知在该节点上监听的下一个客户端，该客户端会重新检查自己是否是最小的序号节点，若是，则获得锁。

ZK 可以实现哪些类型的分布式锁

ZooKeeper 提供了顺序节点和 Watch 机制，使其成为实现分布式锁的理想选择。通过创建临时节点（Ephemeral Node）或临时顺序节点（Ephemeral Sequential Node），可以在分布式系统中实现对共享资源的访问控制。ZooKeeper 可以实现的分布式锁类型不限于：

  • (1) 非公平锁（NonFairLock）

    • 实现步骤：
      • 通过临时节点（没有编号）实现。
      • 每个客户端分别在同一父节点下创建临时节点。
      • 第一个成功创建临时节点的客户端将获得锁，可以执行操作。
      • 其他客户端创建临时节点失败后，需要对该锁节点注册一个删除事件监听器。
      • 一旦当前持锁客户端删除该锁节点（即释放锁）时，监听器会触发，其他客户端就会被通知可以重新尝试创建临时节点以获取锁。
    • 实现特点：
      • 实现简单，但无法保证请求锁的顺序公平性。
      • 后来的客户端有可能比先前失败的客户端先获取到锁（抢锁问题）。
      • 如果有高频率的加锁请求，某些客户端可能一直失败，始终没机会获得锁（饥饿问题）。
    • 适用场景：
      • 对锁获取顺序不敏感、只需要简单互斥的情况。

  • (2) 公平锁（FairLock）

    • 实现步骤：
      • 通过临时顺序节点（有编号）实现。
      • 每个客户端分别在同一父节点下创建临时顺序节点。
      • ZooKeeper 会自动分配递增编号，编号最小的客户端将获得锁，可以执行操作。
      • 其他客户端不会监听父节点的所有变化，只需要监听编号最接近且小于自己的那个节点（即前一个节点）的删除事件。
      • 当锁节点被删除后，负责监听该节点的客户端会收到通知，然后重新判断当前是否满足获取锁的条件，如果满足则立即获得锁，否则继续监听下一个编号最接近且小于自己的节点。
    • 实现特点：
      • 先到先得：保证请求锁的顺序按照创建节点的顺序分配锁，公平性高。
      • 避免抢锁和饥饿问题：长期等待的客户端不会被后来加入的客户端 “插队”，而且每个客户端都有机会获得锁。
      • 避免惊群效应：每个客户端只需要监听前一个节点，而不是监听父节点的所有变化，减少了 ZooKeeper 的事件通知量。
    • 适用场景：
      • 适合对锁获取顺序有严格要求的业务场景，例如：分布式任务队列处理、分布式写操作顺序控制、对公平性要求高的集群资源分配。
      • 不适合锁竞争非常激烈且对延迟敏感的场景，因为顺序锁每次释放都需要重新判断前驱节点，可能会引入一些延迟。

  • (3) 读写锁（ReadWriteLock）

    • 基于临时顺序节点（有编号）并通过节点命名区分读写类型（如 read-0000003 / write-0000005）。
    • 实现读锁：
      • 客户端创建 read- 前缀的临时顺序节点。
      • 检查比自己编号小的节点中是否存在写锁节点（write- 前缀）：
        • 如果没有写锁节点，则立即获得读锁（多个读锁可并行）。
        • 如果有写锁节点，则监听编号最接近且小于自己的那个写锁节点的删除事件。当该节点被删除后，重新判断当前是否满足获取锁的条件，如果满足则立即获得锁，否则继续监听下一个编号最接近且小于自己的写锁节点。
    • 实现写锁：
      • 客户端创建 write- 前缀的临时顺序节点。
      • 检查比自己编号小的所有节点（包括读锁和写锁）是否为空：
        • 如果为空，则获得写锁。
        • 如果不为空，则监听编号最接近且小于自己的那个节点（即前一个节点）的删除事件。当该节点被删除后，重新判断当前是否满足获取锁的条件，如果满足则立即获得锁，否则继续监听下一个编号最接近且小于自己的节点。
    • 实现特点：
      • 读锁之间不互斥，允许多个客户端并发读。
      • 写锁与读锁、写锁之间互斥，保证数据一致性。
      • 顺序编号和 Watch 机制可以保证公平性和低惊群效应（每个客户端只监听前一个节点）。
    • 适用场景：
      • 读多写少的分布式系统：允许多个客户端同时读取共享资源，提高读吞吐量。
      • 分布式缓存：多个节点可并行读取缓存，但写操作需独占缓存。
      • 配置或状态管理：读操作频繁，写操作较少，但需要保证写入时的数据一致性。
      • 分布式数据库：需要对读写操作进行协调，保证顺序性和互斥性。

代码案例

ZK 实现非公平锁

ZooKeeper 实现分布式公平锁时需要注意的地方

  • 死锁问题：锁不能因为意外就变成死锁，所以要用临时节点，客户端连接断开了，锁就自动释放。
  • 锁等待问题：锁有排队的需求（公平锁），所以要用顺序节点。
  • 锁管理问题：一旦锁的持有者释放了锁，需要通知其他的锁竞争者，所以需要用到监听器（Watcher）。
  • 容错处理问题：在实际应用中，需要处理网络中断、会话过期等异常情况，确保锁的机制不会因为这些问题而失效。
  • 性能问题：ZooKeeper 的性能在一定程度上依赖于节点数量和操作频率，因此需要合理设计锁的粒度和使用频率。
  • 避免惊群效应：多个客户端可能同时尝试获取锁，导致频繁地获取子节点列表和设置监视器。比如，有 1000 个锁竞争者，一旦锁释放了，1000 个竞争者就会得到通知，然后判断自己创建的节点是否是最小的序号节点，若是就获取到锁。其它 999 个竞争者会重新注册监听，这就是惊群效应，出点事就会惊动整个羊群。解决方案是每个竞争者只监听序号最接近且小于自己的那个节点（即前一个节点），比如 2 号释放了锁，那么只有 3 号会得到通知。
  • 引入 Maven 依赖
1
2
3
4
5
<dependency>
    <groupId>org.apache.zookeeper</groupId>
    <artifactId>zookeeper</artifactId>
    <version>3.6.3</version>
</dependency>
  • 非公平锁的实现
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
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooDefs;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.data.Stat;

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

/**
 * 基于 ZooKeeper 的临时节点实现分布式锁（非公平锁），直接抢占锁，容易造成 "饥饿问题" 和 "惊群效应"
 * 特别注意：代码缺乏健壮性和可用性（尤其是对网络抖动、可重入、会话过期的处理），仅供参考学习，不适用于生产环境
 */
public class DistributeNonFairLock {

    private ZooKeeper zk;
    private final static String ROOT_LOCK_PATH = "/locks";
    private final static String LOCK_PREFIX = "lock-";
    private final static CountDownLatch CONNECTED_LATCH = new CountDownLatch(1);
    private static final String ADDRESS = "192.168.2.235:2181,192.168.2.235:2182,192.168.2.235:2183";

    /**
     * 私有构造方法
     */
    private DistributeNonFairLock() {
        try {
            this.zk = new ZooKeeper(ADDRESS, 50000, new ConnectWatcher());

            // 阻塞等待 ZK 客户端建立连接
            CONNECTED_LATCH.await();

            // 确保根节点存在
            if (zk.exists(ROOT_LOCK_PATH, false) == null) {
                try {
                    // 创建持久节点
                    zk.create(ROOT_LOCK_PATH, "".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
                } catch (Exception ignored) {

                }
            }

            System.out.println("ZooKeeper server connect success.");
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 获取一个分布式锁
     */
    public boolean tryLock(String lockId, long waitTimeMs) {
        String path = ROOT_LOCK_PATH + "/" + LOCK_PREFIX + lockId;
        try {
            // 尝试创建临时节点（抢占锁），若失败会抛出异常
            zk.create(path, "".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
            System.out.println("acquire the lock for [id=" + lockId + "].");
            return true;
        } catch (Exception e) {
            // 整体处理的超时时间
            long deadline = System.currentTimeMillis() + waitTimeMs;
            while (true) {
                try {
                    // 判断是否有剩余的时间可用
                    long remaining = deadline - System.currentTimeMillis();
                    if (remaining <= 0) {
                        return false;
                    }

                    // 提前创建本轮专属的 Latch（避免错过节点被删除的事件，导致线程永远阻塞）
                    CountDownLatch waitLatch = new CountDownLatch(1);

                    // 判断节点是否存在，并给该节点注册一个监听器，负责监听该节点被删除
                    Stat stat = zk.exists(path, new NodeDeletedWatcher(path, waitLatch));

                    // 如果节点已存在，阻塞等待该节点被删除
                    if (stat != null) {
                        // 双重检查，再次判断节点是否存在，防止线程死等
                        if (zk.exists(path, false) != null) {
                            // 阻塞等待节点被删除
                            boolean signaled = waitLatch.await(remaining, TimeUnit.MILLISECONDS);
                            // 如果阻塞等待超时了，直接放弃继续抢占锁
                            if (!signaled) {
                                return false;
                            }
                        }
                    }

                    // 如果节点不存在或者阻塞等待被唤醒，则再次尝试创建临时节点（抢占锁），若失败会抛出异常
                    zk.create(path, "".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);

                    System.out.println("acquire the lock for [id=" + lockId + "].");
                    return true;
                } catch (Exception ee) {
                    // 节点已存在，继续下一轮抢占锁，当出现网络抖动或者节点刚删除立刻又被别人创建时，会导致 CPU 占用飙升，建议这里睡眠一段时间或者限制重试次数
                    sleepQuiet(50);
                }
            }
        }
    }

    /**
     * 释放掉一个分布式锁
     */
    public void unlock(String lockId) {
        String path = ROOT_LOCK_PATH + "/" + LOCK_PREFIX + lockId;
        try {
            // 删除节点（如果指定的版本为 -1，则它与该节点的任何版本匹配）
            zk.delete(path, -1);
            System.out.println("release the lock for [id=" + lockId + "].");
        } catch (KeeperException.NoNodeException ignored) {
            // 节点不存在是正常情况，可忽略异常信息
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * ZK 连接事件的 Watcher
     */
    static class ConnectWatcher implements Watcher {

        @Override
        public void process(WatchedEvent event) {
            // 处理客户端连接状态事件
            if (Event.KeeperState.SyncConnected == event.getState()) {
                CONNECTED_LATCH.countDown();
            }
        }

    }

    /**
     * ZK 节点被删除事件的 Watcher
     */
    static class NodeDeletedWatcher implements Watcher {

        private final String path;
        private final CountDownLatch latch;

        NodeDeletedWatcher(String path, CountDownLatch latch) {
            this.path = path;
            this.latch = latch;
        }

        @Override
        public void process(WatchedEvent event) {
            // 处理节点被删除的事件
            if (event.getType() == Event.EventType.NodeDeleted && path.equals(event.getPath())) {
                latch.countDown();
            }
        }
    }

    /**
     * 关闭 ZK 连接
     */
    public void close() {
        try {
            if (zk != null) {
                zk.close();
            }
        } catch (Exception ignored) {

        }
    }

    /**
     * 睡眠一段时间
     */
    private void sleepQuiet(long mills) {
        try {
            Thread.sleep(mills);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }

    /**
     * 封装单例的静态内部类
     */
    private static class Singleton {

        private static DistributeNonFairLock instance;

        static {
            instance = new DistributeNonFairLock();
        }

        public static DistributeNonFairLock getInstance() {
            return instance;
        }

    }

    /**
     * 获取单例
     */
    public static DistributeNonFairLock getInstance() {
        return Singleton.getInstance();
    }

}
  • 非公平锁的测试代码
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
public class DistributeLockTest {

    public static void main(String[] args) throws Exception {
        String lockId = UUID.randomUUID().toString();
        DistributeNonFairLock noneFailLock = DistributeNonFairLock.getInstance();

        new Thread(new Runnable() {
            @Override
            public void run() {
                boolean locked = false;
                try {
                    // 抢占锁
                    locked = noneFailLock.tryLock(lockId, 15 * 1000);
                    if (locked) {
                        System.out.println("Thread 1 获取锁成功");
                        Thread.sleep(10 * 1000);
                    } else {
                        System.out.println("Thread 1 获取锁失败");
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                } finally {
                    // 释放锁
                    if (locked) {
                        noneFailLock.unlock(lockId);
                        System.out.println("Thread 1 释放锁");
                    }
                }
            }
        }).start();

        new Thread(new Runnable() {
            @Override
            public void run() {
                boolean locked = false;
                try {
                    // 抢占锁
                    locked = noneFailLock.tryLock(lockId, 15 * 1000);
                    if (locked) {
                        System.out.println("Thread 2 获取锁成功");
                        Thread.sleep(10 * 1000);
                    } else {
                        System.out.println("Thread 2 获取锁失败");
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                } finally {
                    // 释放锁
                    if (locked) {
                        noneFailLock.unlock(lockId);
                        System.out.println("Thread 2 释放锁");
                    }
                }
            }
        }).start();
    }
}
  • 程序运行的结果
1
2
3
4
5
6
7
8
9
ZooKeeper server connect success.
acquire the lock for [id=51a023ce-c6f9-4aee-9309-de234cb869fb].
Thread 1 获取锁成功
release the lock for [id=51a023ce-c6f9-4aee-9309-de234cb869fb].
Thread 1 释放锁
acquire the lock for [id=51a023ce-c6f9-4aee-9309-de234cb869fb].
Thread 2 获取锁成功
release the lock for [id=51a023ce-c6f9-4aee-9309-de234cb869fb].
Thread 2 释放锁

ZK 实现公平锁

  • 引入 Maven 依赖
1
2
3
4
5
<dependency>
    <groupId>org.apache.zookeeper</groupId>
    <artifactId>zookeeper</artifactId>
    <version>3.6.3</version>
</dependency>
  • 公平锁的实现
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
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooDefs;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.data.Stat;

import java.util.Collections;
import java.util.List;
import java.util.concurrent.CountDownLatch;

/**
 * 基于 ZooKeeper 临时顺序节点实现分布式锁（公平锁），先到先得，可以避免 "饥饿问题" 和 "惊群效应"
 * 特别注意：代码缺乏健壮性和可用性（尤其是对网络抖动、可重入、会话过期的处理），仅供参考学习，不适用于生产环境
 */
public class DistributeFairLock {

    // ZooKeeper 连接地址（多个集群节点使用逗号分割）
    private static final String ADDRESS = "192.168.2.235:2181,192.168.2.235:2182,192.168.2.235:2183";

    // ZooKeeper 会话超时时间
    private static final int SESSION_TIMEOUT = 2000;

    // 根节点的路径
    private static final String ROOT_PATH = "/locks";

    // 子节点的路径前缀
    private static final String PREFIX_CHILD_PATH = "seq-";

    // ZooKeeper 客户端
    private ZooKeeper client;

    // 当前客户端创建的子节点的路径
    private String currentChildNodePath;

    // 当前客户端监听的前一个子节点的路径
    private String preChildNodePath;

    // 等待客户端建立连接的锁
    private CountDownLatch connectLatch = new CountDownLatch(1);

    public DistributeFairLock() throws Exception {
        // 初始化客户端
        client = new ZooKeeper(ADDRESS, SESSION_TIMEOUT, new ConnectWatcher());

        // 阻塞等待客户端建立连接
        connectLatch.await();

        // 如果根节点不存在，则创建根节点，根节点类型为永久节点
        if (client.exists(ROOT_PATH, false) == null) {
            client.create(ROOT_PATH, "locks".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
        }

        System.out.println("ZooKeeper server connect success.");
    }

    /**
     * 加锁
     */
    public void lock() throws Exception {
        // 创建当前子节点（子节点的类型为临时顺序节点）
        currentChildNodePath = client.create(ROOT_PATH + "/" + PREFIX_CHILD_PATH, null, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);

        // 获取所有子节点
        List<String> children = client.getChildren(ROOT_PATH, false);

        // 如果只有一个子节点，则当前客户端直接获得锁
        if (children.size() == 1) {
            return;
        }

        // 子节点根据序号进行排序
        Collections.sort(children);

        // 获取当前子节点的名称，如：seq-0000000009
        String currentChildNodeName = currentChildNodePath.substring((ROOT_PATH + "/").length());

        // 通过子节点名称，获取当前子节点在子节点集合中的位置
        int currentChildNodeIndex = children.indexOf(currentChildNodeName);

        // 子节点位置不正常，直接抛出异常
        if (currentChildNodeIndex == -1) {
            throw new RuntimeException("Current child node not in children list.");
        }
        // 如果当前子节点是第一个子节点，则当前客户端直接获得锁
        else if (currentChildNodeIndex == 0) {
            System.out.println("acquire the lock.");
            return;
        }
        // 如果当前子节点不是第一个子节点，则监听前一个子节点被删除
        else {
            // 获取前一个子节点的路径
            preChildNodePath = ROOT_PATH + "/" + children.get(currentChildNodeIndex - 1);

            // 新建一个专用的 Latch
            CountDownLatch watchPreNodeLatch = new CountDownLatch(1);

            // 判断前一个子节点是否存在，并给该节点注册一个监听器，负责监听该节点被删除
            Stat stat = client.exists(preChildNodePath, new NodeDeletedWatcher(preChildNodePath, watchPreNodeLatch));

            // 如果前一个子节点已存在，阻塞等待该节点被删除（进入等待锁状态）
            if (stat != null) {
                // 双重检查，再次判断前一个子节点是否存在，防止线程死等
                if (client.exists(preChildNodePath, false) != null) {
                    // 阻塞等待前一个子节点被删除
                    watchPreNodeLatch.await();
                }
            }

            // 如果前一个子节点不存在或者阻塞等待被唤醒，则当前客户端直接获得锁
            System.out.println("acquire the lock.");
            return;
        }
    }

    /**
     * 释放锁
     */
    public void unlock() {
        try {
            // 删除子节点（如果指定的版本为 -1，则它与该节点的任何版本匹配）
            client.delete(currentChildNodePath, -1);
            System.out.println("release the lock.");
        } catch (KeeperException.NoNodeException ignored) {
            // 子节点不存在是正常情况，可忽略异常信息
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * ZK 连接事件的 Watcher
     */
    class ConnectWatcher implements Watcher {

        @Override
        public void process(WatchedEvent event) {
            // 处理客户端连接状态事件
            if (Event.KeeperState.SyncConnected == event.getState()) {
                connectLatch.countDown();
            }
        }

    }

    /**
     * ZK 节点被删除事件的 Watcher
     */
    class NodeDeletedWatcher implements Watcher {

        private final String path;
        private final CountDownLatch latch;

        NodeDeletedWatcher(String path, CountDownLatch latch) {
            this.path = path;
            this.latch = latch;
        }

        @Override
        public void process(WatchedEvent event) {
            // 处理节点被删除的事件（释放锁）
            if (event.getType() == Event.EventType.NodeDeleted && path.equals(event.getPath())) {
                latch.countDown();
            }
        }
    }

    /**
     * 关闭 ZK 连接
     */
    public void close() {
        try {
            if (client != null) {
                client.close();
            }
        } catch (Exception ignored) {

        }
    }

}
  • 公平锁的测试代码
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
public class DistributeLockTest {

    public static void main(String[] args) throws Exception {
        DistributeFairLock lock1 = new DistributeFairLock();
        DistributeFairLock lock2 = new DistributeFairLock();

        new Thread(new Runnable() {
            @Override
            public void run() {
                try {
                    // 抢占锁
                    lock1.lock();
                    System.out.println("Thread 1 获得锁");
                    Thread.sleep(10 * 1000);
                } catch (Exception e) {
                    e.printStackTrace();
                } finally {
                    // 释放锁
                    lock1.unlock();
                    System.out.println("Thread 1 释放锁");
                }
            }
        }).start();

        new Thread(new Runnable() {
            @Override
            public void run() {
                try {
                    // 抢占锁
                    lock2.lock();
                    System.out.println("Thread 2 获得锁");
                    Thread.sleep(10 * 1000);
                } catch (Exception e) {
                    e.printStackTrace();
                } finally {
                    // 释放锁
                    lock2.unlock();
                    System.out.println("Thread 2 释放锁");
                }
            }
        }).start();
    }

}
  • 程序运行的结果
1
2
3
4
5
6
7
8
9
ZooKeeper server connect success.
acquire the lock.
Thread 1 获得锁
release the lock.
Thread 1 释放锁
acquire the lock.
Thread 2 获得锁
release the lock.
Thread 2 释放锁

代码下载

本文所需的完整案例代码，可以直接从 GitHub 下载对应章节 distributed-locks-05。