Zookeeper Java 开发，自定义分布式锁示例

如果您还没有安装Zookeeper请看ZooKeeper 安装说明，Zookeeper 命令使用方法和数据说明，Zookeeper Java 开发入门。

一、概述

• 情景：假设有10个客户端（分散的10台主机）要执行一个任务，这个任务某些过程需要保持原子性。那么我们就需要一个分布式锁。

• 原理：通过在Zookeeper中创建序列节点来实现获得锁，删除节点来释放锁。其实质是一个按先来后到的排序过程，实现过程如下：

  • 客户端发起请求，创建锁序列节点（/lock/xxxxxxxx）

  • 获取所有锁节点，判断自己是否为最小节点

    • 如果自己是最小序列节点，则立即获得锁。
    • 否则不能获得锁，但要监控前一个序列节点的状态。

  • 获得锁的客户端开始执行任务。

  • 执行完任务后释放锁。

    • 由于后一个节点监控了前一个节点，当前一个节点删除时，后一个客户端会收到回调。

    • 在这个回调中再获取所有锁节点，判断自己是否为最小节点。

    • 以此类推，直到全部结束。

• 流程如下

• 如果您对没有做过 Zookeeper 开发，强烈建立先看 Zookeeper Java 开发入门。

二、导入依赖包

• 在 pom.xml 文件中导入 Zookeeper 包，注意一般这个包的版本要和您安装的 Zookeeper 服务端版本一致。

  <dependency>
     <groupId>org.apache.zookeepergroupId>
      <artifactId>zookeeperartifactId>
      <version>3.8.2version>
  dependency>
  1
  2
  3
  4
  5

三、创建锁的过程

3.1 通过 create 创建节点信息

• 通过 create 创建序列节点信息。他是异步方式，创建成功后会调用 AsyncCallback.StringCallback.processResult 回调函数。

    public void lock() throws InterruptedException, LockException {
        zooKeeper.create("/lock", "xxx".getBytes(),
                ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL, this, context);
        countDownLatch.await();

        if(StringUtils.isEmpty(this.lockNodePath)){
            throw new LockException("创建锁失败");
        }

        System.out.println(this.appName + " 获得锁");
    }
1
2
3
4
5
6
7
8
9
10
11

3.2 AsyncCallback.StringCallback 回调函数

• 在 AsyncCallback.StringCallback 的回调函数中通过 getChildren 方法获取 ZooKeeper 锁节点下的所有节点信息。这个方法是异步的，调用成功后会调用 AsyncCallback.Children2Callback.processResult 回调函数。

    // AsyncCallback.StringCallback
    @Override
    public void processResult(int i, String s, Object o, String s1) {
        if(StringUtils.isEmpty(s1)){
            // 这里是创建锁失败的情况。
            this.countDownLatch.countDown();
            return;
        }
        System.out.println(this.appName + " create lock node="+s1);

        this.lockNodePath = s1;
        // 获取 ZooKeeper 锁节点下的所有节点信息，以此来判断我是不是第一个创建节点，如果就获得锁，否则监控前一个节点信息。
        zooKeeper.getChildren("/", false, this, context);
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14

3.3 AsyncCallback.Children2Callback 的回调函数

• 在 AsyncCallback.Children2Callback 的回调函数判断我是不是第一个创建节点，如果就获得锁，否则监控前一个节点信息。监控前一个节点信息使用 exists 方法，这个方法设置了 Watcher 的 processResult 回调函数

    // AsyncCallback.Children2Callback
    @Override
    public void processResult(int i, String s, Object o, List<String> list, Stat stat) {

        Collections.sort(list);

//        for (String s1 : list) {
//            System.out.println("\t "+this.lockNodePath+" previous lock node="+s1);
//        }

        int index = list.indexOf(lockNodePath.substring(1));
        if(0 == index){
            // 如果我现在是第一个节点，则获得锁
            try {
                zooKeeper.setData("/", this.lockNodePath.getBytes(), -1);
            } catch (KeeperException e) {
                e.printStackTrace();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            this.countDownLatch.countDown();
        }
        else {
            // 我不是第一个节点，监控前一个节点信息（等他删除后我就是可能是第一个了）
            String watchNodePath = "/" + list.get(index - 1);
            System.out.println("\t "+this.lockNodePath+" watch node："+ watchNodePath);
            zooKeeper.exists(watchNodePath, this, new StatCallback() {
                @Override
                public void processResult(int i, String s, Object o, Stat stat) {

                }
            }, context);
        }
    }
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

3.4 Watcher 的回调函数

• 在 Watcher 的回调函数，我们通过判断 watchedEvent.getType() 为 NodeDeleted 类型时，重新获取 ZooKeeper 锁节点下的所有节点信息，这使得消息回到了 “3.3”步，判断谁是第一个节点，然后获得得，完成整个流程。

    // Watcher
    @Override
    public void process(WatchedEvent watchedEvent) {
        switch (watchedEvent.getType()) {
            case None:
                break;
            case NodeCreated:
                break;
            case NodeDeleted:
                zooKeeper.getChildren("/", false, this, context);
                break;
            case NodeDataChanged:
                break;
            case NodeChildrenChanged:
                break;
            case DataWatchRemoved:
                break;
            case ChildWatchRemoved:
                break;
            case PersistentWatchRemoved:
                break;
        }
    }

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

四、完整示例

4.1 完整分布式锁代码

package top.yiqifu.study.p131;


import org.apache.commons.lang3.StringUtils;
import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;

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

public class ZookeeperLock implements Watcher, AsyncCallback.StringCallback,
        AsyncCallback.Children2Callback {

    private String appName;
    private ZooKeeper zooKeeper;

    private Object context;
    private String lockNodePath;
    private CountDownLatch countDownLatch = new CountDownLatch(1);

    public ZookeeperLock(String name, ZooKeeper zk){
        this.appName = name;
        this.zooKeeper = zk;
        this.context = this;
    }


    public void lock() throws InterruptedException, LockException {
        zooKeeper.create("/lock", "xxx".getBytes(),
                ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL, this, context);
        countDownLatch.await();

        if(StringUtils.isEmpty(this.lockNodePath)){
            throw new LockException("创建锁失败");
        }

        System.out.println(this.appName + " 获得锁");
    }

    public void unlock() throws KeeperException, InterruptedException, LockException {
        if(StringUtils.isEmpty(this.lockNodePath)){
            throw new LockException("没有获得锁，无法释放锁");
        }
        zooKeeper.delete(lockNodePath, -1);

        System.out.println(this.appName + " 释放锁");
    }


    // AsyncCallback.StringCallback
    @Override
    public void processResult(int i, String s, Object o, String s1) {
        if(StringUtils.isEmpty(s1)){
            // 这里是创建锁失败的情况。
            this.countDownLatch.countDown();
            return;
        }
        System.out.println(this.appName + " create lock node="+s1);

        this.lockNodePath = s1;
        // 获取 ZooKeeper 锁节点下的所有节点信息，以此来判断我是不是第一个创建节点，如果就获得锁，否则监控前一个节点信息。
        zooKeeper.getChildren("/", false, this, context);
    }

    // AsyncCallback.Children2Callback
    @Override
    public void processResult(int i, String s, Object o, List<String> list, Stat stat) {

        Collections.sort(list);

//        for (String s1 : list) {
//            System.out.println("\t "+this.lockNodePath+" previous lock node="+s1);
//        }

        int index = list.indexOf(lockNodePath.substring(1));
        if(0 == index){
            // 如果我现在是第一个节点，则获得锁
            try {
                zooKeeper.setData("/", this.lockNodePath.getBytes(), -1);
            } catch (KeeperException e) {
                e.printStackTrace();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            this.countDownLatch.countDown();
        }
        else {
            // 我不是第一个节点，监控前一个节点信息（等他删除后我就是可能是第一个了）
            String watchNodePath = "/" + list.get(index - 1);
            System.out.println("\t "+this.lockNodePath+" watch node："+ watchNodePath);
            zooKeeper.exists(watchNodePath, this, new StatCallback() {
                @Override
                public void processResult(int i, String s, Object o, Stat stat) {

                }
            }, context);
        }
    }

    // Watcher
    @Override
    public void process(WatchedEvent watchedEvent) {
        switch (watchedEvent.getType()) {
            case None:
                break;
            case NodeCreated:
                break;
            case NodeDeleted:
                zooKeeper.getChildren("/", false, this, context);
                break;
            case NodeDataChanged:
                break;
            case NodeChildrenChanged:
                break;
            case DataWatchRemoved:
                break;
            case ChildWatchRemoved:
                break;
            case PersistentWatchRemoved:
                break;
        }
    }



    public class LockException extends  Exception
    {
        public LockException(String message){
            super(message);
        }
    }
}

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

4.2 测试类

package top.yiqifu.study.p131;


import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;

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

public class Test06_ZookeeperLock {

    public static void main(String[] args) {
        try {
            // 创建 ZooKeeper 对象
            final ZooKeeper zooKeeper = testCreateZookeeper();

            int clientCount = 10;
            final CountDownLatch countDownLatch = new CountDownLatch(clientCount);
            for (int i = 0; i < clientCount; i++) {
                new Thread(new Runnable(){
                    @Override
                    public void run() {
                        TestLock(zooKeeper);
                        countDownLatch.countDown();
                    }
                }).start();
            }

            countDownLatch.await();
        } catch (IOException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (KeeperException e) {
            e.printStackTrace();
        }
    }

    private static void  TestLock(ZooKeeper zooKeeper){
        try {
            String appName = Thread.currentThread().getName();
            ZookeeperLock zookeeperLock = new ZookeeperLock(appName, zooKeeper);
            
            // 加锁（获得分布式锁）
            zookeeperLock.lock();
            System.out.println(appName + " 执行任务");
            Thread.sleep(1000);
		   // 释放锁
            zookeeperLock.unlock();

        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (KeeperException e) {
            e.printStackTrace();
        } catch (ZookeeperLock.LockException e) {
            e.printStackTrace();
        }
    }

    private static ZooKeeper testCreateZookeeper() throws IOException, InterruptedException, KeeperException {
        final CountDownLatch countDownLatch = new CountDownLatch(1);

        // ZooKeeper 集群地址（没连接池的概念，是Session的概念）
        //String connectionString = "192.168.8.51:2181,192.168.8.52:2181,192.168.8.53:2181";
        String connectionString = "192.168.8.51:2181,192.168.8.52:2181,192.168.8.53:2181/aaa";
        // ZooKeeper Session 数据超时时间（也就是这个Session关闭后，与这个Session相关的数据在ZooKeeper中保存多信）。
        Integer sessionTimeout = 3000;
        // ZooKeeper Session 级别 Watcher（Watch只发生在读方法上，如 get、exists）
        final ZooKeeper zooKeeper = new ZooKeeper(connectionString, sessionTimeout, new Watcher() {
            @Override
            public void process(WatchedEvent watchedEvent) {
                try {
                    Event.KeeperState state = watchedEvent.getState();
                    Event.EventType type = watchedEvent.getType();
                    String path = watchedEvent.getPath();

                    switch (state) {
                        case Unknown:
                            break;
                        case Disconnected:
                            break;
                        case NoSyncConnected:
                            break;
                        case SyncConnected:
                            countDownLatch.countDown();
                            break;
                        case AuthFailed:
                            break;
                        case ConnectedReadOnly:
                            break;
                        case SaslAuthenticated:
                            break;
                        case Expired:
                            break;
                        case Closed:
                            break;
                    }
                    switch (type) {
                        case None:
                            break;
                        case NodeCreated:
                            break;
                        case NodeDeleted:
                            break;
                        case NodeDataChanged:
                            break;
                        case NodeChildrenChanged:
                            break;
                        case DataWatchRemoved:
                            break;
                        case ChildWatchRemoved:
                            break;
                        case PersistentWatchRemoved:
                            break;
                    }

                    System.out.println("Session watch state=" + state);
                    System.out.println("Session watch type=" + type);
                    System.out.println("Session watch path=" + path);
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        });

        countDownLatch.await();
        ZooKeeper.States state = zooKeeper.getState();
        switch (state) {
            case CONNECTING:
                break;
            case ASSOCIATING:
                break;
            case CONNECTED:
                break;
            case CONNECTEDREADONLY:
                break;
            case CLOSED:
                break;
            case AUTH_FAILED:
                break;
            case NOT_CONNECTED:
                break;
        }
        System.out.println("ZooKeeper state=" + state);

        return zooKeeper;
    }


}

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