zookeeper分布式锁实现

通过微信公众号查看本文，效果更明显哦，文章链接：https://mp.weixin.qq.com/s/VeSCrZ-dlPdLKGTY2xoZ7w

一、zookeeper介绍
zookeeper是一个分布式的、开放源码的分布式应用程序协调服务，是Hadoop和Hbase的重要组件。在zook中，znode是一个跟Unix文件系统路径相似的节点，可以往这个节点存储或获取数据，通过客户端可对znode进行增删查改操作，还可以注册watcher监控znode的变化。
Zookeeper节点类型：持久节点、持久顺序节点、临时节点、临时顺序节点
对于持久节点和临时节点，同一个znode下，节点名称是唯一的，这是实现分布式锁的基础。
二、单服务并发锁
（1）不用锁：使用CountDownLatch 模拟10个并发线程，在不使用锁的情况下，运行下面代码

public class ConcurrentTest {
    public static void main(String[] args) {
        CountDownLatch countDownLatch = new CountDownLatch(10);
        for (int i = 0; i < 10; i++){
            new Thread(new Runnable() {
                @Override
                public void run() {
                    TestService testService = new TestService();
                    countDownLatch.countDown();
                    try {
                        countDownLatch.await();
                        testService.consoleInfo();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }).start();
        }
    }
    static class TestService{
        private static OrderCodeGenerator ocg = new OrderCodeGenerator();
        public void consoleInfo(){
            System.out.println(ocg.getOrderCode());
        }
    }
}
public class OrderCodeGenerator {
    private static int i = 0;
    public String getOrderCode(){
        return "order:" + i++;
    }
}

可观察在不适用锁的情况下，结果并不是线程安全的，结果如下（可运行多次进行比较）：
order:0
order:6
order:5
order:4
order:3
order:2
order:0
order:0
order:0
order:1

（2）synchronized加锁，使用synchronized (this)对代码进行加锁，然后运行上述代码：

static class TestService{
    private static OrderCodeGenerator ocg = new OrderCodeGenerator();
    public void consoleInfo(){
        synchronized (this){
            System.out.println(ocg.getOrderCode());
        }
    }
}

结果如下：从结果可知，使用synchronized(this)也没有保证线程安全，因为this代表的是for循环中每个线程自己new出来的对象（TestService testService = new TestService()），并不是所有线程指定的同一个对象，所以并不能保险线程安全。
order:0
order:6
order:7
order:5
order:1
order:1
order:4
order:2
order:0
order:3
将上述的synchronized(this)修改成synchronized (TestService.class)或者synchronized(ocg )后在运行代码，会发现无论运行多少次，都是线程安全的;
order:0
order:1
order:2
order:3
order:4
order:5
order:6
order:7
order:8
order:9

（3）Lock加锁

static class TestService{
    private static OrderCodeGenerator ocg = new OrderCodeGenerator();
    /** 定义成静态锁，确保每个线程进来后都是使用同一个锁 */
    private static Lock lock = new ReentrantLock();
    public void consoleInfo(){
        try{
            lock.lock();
            System.out.println(ocg.getOrderCode());
        }catch (Exception e){
            e.printStackTrace();
        }finally {
            lock.unlock();
        }
    }
}

结果显示线程安全：
order:0
order:1
order:2
order:3
order:4
order:5
order:6
order:7
order:8
order:9

三、zookeeper分布式锁
原理：利用zookeeper同一个节点名称不能重复的原理来实现分布式锁；
加锁：创建指定名称的节点，如果能创建成功，则获得锁（加锁成功），如果节点已存在，就标志锁已被别人获取，此时就得等待别人释放锁；
释放锁：删除指定名称的节点
缺点：客户端无故接受了很多与自己无关时间的通知； 存在死锁的可能性（如果节点没被删掉就会出现死锁）
（1）zookeeper安装测试：
http://archive.apache.org/dist/zookeeper/zookeeper-3.3.6/zookeeper-3.3.6.tar.gz
下载完成后解压到G盘：G:\zookeeper

进入G:\zookeeper\conf目录下，复制zoo_sample.cfg重命名zoo.cfg，并修改zoo.cfg中的dataDir和dataDirLog配置，并在相应目录下新建号data和log文件

# The number of milliseconds of each tick
tickTime=2000
# The number of ticks that the initial
# synchronization phase can take
initLimit=10
# The number of ticks that can pass between
# sending a request and getting an acknowledgement
syncLimit=5
# the directory where the snapshot is stored.
dataDir=G:\\zookeeper\\data 
dataDirLog=G:\\zookeeper\\log
# the port at which the clients will connect
clientPort=2181

启动zookeeper服务端，执行下列命令：
cd G:/zookeeper/bin
ZkServer.cmd


启动成功后，新开一个cmd窗口启动客户端，执行命令 zkCli.cmd

然后可以对zookeeper进行操作
创建节点名称（命令 节点名称 值）：create /test hello world
修改节点值：set /test hello zookeeper
获取节点值：get /test   输出结果：hello zookeeper
删除节点：delete /test
创建临时节点： create -e /test hello world
创建临时顺序节点：create -e -s /test hello world

（2）通过创建持久性节点加锁，原理在上面已有介绍，直接上代码
/**添加pom依赖：*/

<dependency>
    <groupId>org.apache.zookeeper</groupId>
    <artifactId>zookeeper</artifactId>
    <version>3.4.9</version>
</dependency>
<dependency>
    <groupId>com.github.sgroschupf</groupId>
    <artifactId>zkclient</artifactId>
    <version>0.1</version>
</dependency>

代码实现：

public class ZkLockUtil implements Lock{
    /** 节点名称 */
    private String nodeName;
    /** zookeeper客户端 */
    private ZkClient zkClient;
    public ZkLockUtil(String nodeName) {
        zkClient = new ZkClient("localhost:2181");
        zkClient.setZkSerializer(new MyZkSerializer());
        this.nodeName = nodeName;
    }
    @Override
    public void lock() {
        if (!tryLock()){
            //等待锁
            this.waitForLock();
            //再次尝试获取锁
            lock();
        }
    }
    /**
     * 监听nodeName节点是否被删除，如果被删除，尝试加锁，如果节点还存在，则继续监听，直到监听到节点被删除
     * 通过countDownLatch让自己阻塞，等待线程唤醒
     */
    private void waitForLock(){
        CountDownLatch countDownLatch = new CountDownLatch(1);
        IZkDataListener listener = new IZkDataListener() {
            @Override
            public void handleDataChange(String s, Object o) throws Exception {
                System.out.println("监听到节点：" + s + "，数据被改变成：" + o);
            }
            @Override
            public void handleDataDeleted(String s) throws Exception {
                System.out.println("监听到节点" + s + "已被删除");
                countDownLatch.countDown();
            }
        };
        //将listener注册到zkClient中
        this.zkClient.subscribeDataChanges(nodeName, listener);
        //如果nodeName仍旧存在，则继续监听，让当前线程阻塞
        if (this.zkClient.exists(nodeName)){
            try{
                countDownLatch.await();
            }catch (Exception e){
            }
        }
        //监听到节点已被删除，将listener注销掉
        this.zkClient.unsubscribeDataChanges(nodeName, listener);
    }
    @Override
    public boolean tryLock() {
        try{
            //创建节点
            this.zkClient.createPersistent(nodeName);
        }catch (Exception e){
            return false;
        }
        return true;
    }
    @Override
    public void unlock() {
        zkClient.delete(nodeName);
    }
   //省略Lock的其余实现方法
    ...
}

修改TestService类的代码：

static class TestService{
    private static OrderCodeGenerator ocg = new OrderCodeGenerator();
    /** 定义成静态锁，确保每个线程进来后都是使用同一个锁 */
    private static Lock lock = new ZkLockUtil("/test");
    public void consoleInfo(){
        try{
           lock.lock();
            System.out.println(Thread.currentThread().getName() + "==========" +ocg.getOrderCode());
        }catch (Exception e){
           e.printStackTrace();
        }finally {
            lock.unlock();
        }
    }
}

运行ConcurrentTest类的main方法：结果显示是线程安全的
Thread-4==========order:0
Thread-7==========order:1
Thread-1==========order:2
Thread-3==========order:3
Thread-9==========order:4
Thread-5==========order:5
Thread-2==========order:6
Thread-0==========order:7
Thread-8==========order:8
Thread-6==========order:9
除了上述打印之外，还打印出了下列的日志：
“监听到节点/test已被删除”
并且打印次数多余10次，在此就显示了其缺点之一：客户端无故接受了很多与自己无关时间的通知
（3）创建临时顺序节点实现分布式锁

原理：利用zookeeper同父子节点不可重名的特点，创建临时顺序节点实现分布式锁，在一些列的顺序节点中，按照节点顺序依次加锁、释放锁
加锁：创建持久性指定父节点，尝试获取锁 --> 如果没有锁，则创建临时顺序节点 --> 获取当前父节点下的临时节点列表 --> 判断当前临时节点是否是序号最小的 --> 如果是，占用锁，执行后续代码 --> 删除当前节点释放锁 --> 如果不是序号最小 --> 对 比自己小的前一个节点 进行监听注册watch --> 当前线程等待锁 --> 对前一节点尝试获取锁 --> 获取子节点列表，判断前一节点是否是最小 --> 如果不是，继续循环上述步骤， 直到将子节点列表中最小的那个节点进行加锁为止
释放锁：删掉顺序节点

代码实现：修改ZkLockUtil类代码

public class ZkLockUtil implements Lock{
    /** 父节点名称 */
    private String nodeName;
    /** zookeeper客户端 */
    private ZkClient zkClient;
    /** 当前节点名称 */
    private ThreadLocal<String> currentNode = new ThreadLocal<>();
    /** 前一节点名称 */
    private ThreadLocal<String> previousNode = new ThreadLocal<>();
    /**
    * 初始化zkClient 如果父节点不存在，则创建父节点
    * @param nodeName
    */
    public ZkLockUtil(String nodeName) {
        zkClient = new ZkClient("localhost:2181");
        zkClient.setZkSerializer(new MyZkSerializer());
        this.nodeName = nodeName;
        if (!this.zkClient.exists(nodeName)){
            this.zkClient.createPersistent(nodeName);
        }
    }
    @Override
    public void lock() {
        if (!tryLock()){
            //等待锁
            this.waitForLock();
            //再次尝试获取锁
            lock();
        }
    }
    @Override
    public void unlock() {
        this.zkClient.delete(this.currentNode.get());
    }
    /**
     * 监听nodeName节点是否被删除，如果被删除，尝试加锁，如果节点还存在，则     
     * 继续监听，直到监听到节点被删除
     * 通过countDownLatch让自己阻塞，等待线程唤醒
     */
    private void waitForLock(){
        CountDownLatch countDownLatch = new CountDownLatch(1);
        IZkDataListener listener = new IZkDataListener() {
            @Override
            public void handleDataChange(String s, Object o) throws Exception {
                System.out.println("监听到节点：" + s + "，数据被改变成：" + o);
            }
            @Override
            public void handleDataDeleted(String s) throws Exception {
                System.out.println("监听到节点" + s + "已被删除");
                countDownLatch.countDown();
            }
        };
        //将listener注册到zkClient中
        this.zkClient.subscribeDataChanges(this.previousNode.get(), listener);
        //如果nodeName仍旧存在，则继续监听，让当前线程阻塞
        if (this.zkClient.exists(this.previousNode.get())){
            try{
                countDownLatch.await();
            }catch (Exception e){
                e.printStackTrace();
            }
        }
        //监听到节点已被删除，将listener注销掉
        this.zkClient.unsubscribeDataChanges(this.previousNode.get(), listener);
    }
    @Override
    public boolean tryLock() {
        //创建临时顺序节点
        if (this.currentNode.get() == null){
          this.currentNode.set(this.zkClient.createEphemeralSequential(nodeName.concat("/"), "testZk"));
        }
        List<String> childNodeList = this.zkClient.getChildren(nodeName);
        Collections.sort(childNodeList);
        //如果当前节点是最小节点，则成功获取锁
        if(this.currentNode.get().equals(nodeName.concat("/").concat(childNodeList.get(0)))){
            return true;
        }
        //获取前一节点
        int currentIndex = childNodeList.indexOf(this.currentNode.get().substring(nodeName.length() + 1));
        this.previousNode.set(nodeName.concat("/").concat(childNodeList.get(currentIndex - 1)));
        return true;
    }
    //省略Lock其他方法...
}

运行ConcurrentTest的测试类可见：结果中每个线程只会监听到一条通知，并且也能实现线程安全。
Thread-5==========order:0
监听到节点/test/0000000000已被删除
Thread-2==========order:1
监听到节点/test/0000000001已被删除
Thread-6==========order:2
监听到节点/test/000000002已被删除
Thread-3==========order:3
监听到节点/test/0000000003已被删除
Thread-8==========order:4
监听到节点/test/0000000007已被删除
Thread-4==========order:5
监听到节点/test/0000000008已被删除
Thread-0==========order:6
监听到节点/test/0000000004已被删除
Thread-9==========order:7
监听到节点/test/0000000009已被删除
Thread-1==========order:8
监听到节点/test/0000000005已被删除
Thread-7==========order:9
监听到节点/test/0000000006已被删除