使用Zookeeper实现分布式锁----基础版

一、思路

锁的原语一般有两个：
①加锁
②解锁

加锁：

加锁的一般算法步骤是：
1.尝试加锁
2.如果锁没有被占用，则加锁成功
3.如果锁被占用，则等待锁被释放
4.锁被释放后，收到锁释放通知，重复步骤 1

翻译成ZooKeeper的算法步骤就是：
1.尝试创建表示锁的临时节点
2.如果创建节点成功，则加锁成功
3.如果创建节点失败，则创建一个锁节点的监视器，等待锁节点的删除通知
4.锁节点被持有者删除后，收到锁节点的删除通知，重复步骤 1

解锁：

解锁的算法步骤是：
1.锁节点是否存在
2.如果不存在，完成解锁
3.如果锁节点存在，则判断锁节点的数据（锁的持有者）是否和解锁者相同
4.如果一样，则删除锁节点，完成解锁

关注公众号： “Java不睡觉”，回复：“资源”。获取大数据全套视频和大量Java书籍

所以根据以上思路，我们写如下的示例程序：

二、代码

2.1 锁实体类

package com.zhb.zookeeper.lock;

public class LockInfo {
    // 锁的名字，体现为zookeeper上的节点名
    String lockname;
    // 锁的持有者，体现为zookeeper锁节点的数据
    String lockOwner;

    public LockInfo(String lockname, String lockOwner) {
        this.lockname = lockname;
        this.lockOwner = lockOwner;
    }

    public String getLockname() {
        return lockname;
    }

    public void setLockname(String lockname) {
        this.lockname = lockname;
    }

    public String getLockOwner() {
        return lockOwner;
    }

    public void setLockOwner(String lockOwner) {
        this.lockOwner = lockOwner;
    }

    @Override
    public String toString() {
        return "LockInfo [lockname=" + lockname + ", lockOwner=" + lockOwner + "]";
    }
}

2.2 锁接口


package com.zhb.zookeeper.lock;

import org.apache.zookeeper.KeeperException;

public interface MyLock {
    /**
     * 尝试加锁,对应加锁步骤1
     * 
     * @return whether success or not, true: lock success, false: lock failed
     */
    public boolean tryLock() throws KeeperException, InterruptedException;

    /**
     * 同步加锁，对应加锁步骤3
     */
    public void lock()throws KeeperException, InterruptedException;

    /**
     * 解锁，释放锁
     */
    public void unlock();
}

2.3 锁的实现类

package com.zhb.zookeeper.lock;

import java.io.IOException;
import java.io.UnsupportedEncodingException;
import java.util.concurrent.CountDownLatch;

import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.Watcher.Event.EventType;
import org.apache.zookeeper.Watcher.Event.KeeperState;
import org.apache.zookeeper.ZooDefs.Ids;
import org.apache.zookeeper.ZooKeeper;

public class ZkLock implements MyLock {

    private String LockName = "dis_lock";
    private LockInfo lockInfo;
    // 此变量的目的是确保连接上zookeeper Server
    private CountDownLatch latch = new CountDownLatch(1);
    private String connectString = "127.0.0.1:2181";
    private int sessionTimeout = 60 * 1000;

    // zookeeperk客户端
    private ZooKeeper zooKeeper = new ZooKeeper(connectString, sessionTimeout, new Watcher() {

        @Override
        public void process(WatchedEvent event) {
            if (KeeperState.SyncConnected == event.getState() && EventType.None == event.getType()) {
                latch.countDown();
            }
        }
    });

    // 构造函数
    public ZkLock(LockInfo lockInfo) throws IOException, InterruptedException {
        this.lockInfo = lockInfo;
        latch.await();
    }

    @Override
    public boolean tryLock() {

        try {
            if (zooKeeper.exists(lockInfo.getLockname(), false) != null) {

                String currentOwner = new String(zooKeeper.getData(lockInfo.getLockname(), false, null));
                System.out.println("已经被加上锁了，锁的持有者是：" + currentOwner);
                return false;
            } else {
                zooKeeper.create(lockInfo.lockname, lockInfo.lockOwner.getBytes(), Ids.OPEN_ACL_UNSAFE,
                        CreateMode.EPHEMERAL);
                return true;
            }

        } catch (KeeperException e) {

        } catch (InterruptedException e) {
            e.printStackTrace();
            System.out.println("####################222222222222222####################");
        }

        return false;
    }

    @Override
    public void lock() {

        if (!tryLock()) {
            CountDownLatch releaseSignal = new CountDownLatch(1);
            try {
                zooKeeper.exists(lockInfo.lockname, new Watcher() {

                    @Override
                    public void process(WatchedEvent event) {
                        if (lockInfo.getLockname().equals(event.getPath())
                                && EventType.NodeDeleted.equals(event.getType())) {
                            releaseSignal.countDown();
                        }
                    }
                });
                releaseSignal.await();
                // 递归调用自己
                lock();
            } catch (KeeperException e1) {
                // TODO Auto-generated catch block
                e1.printStackTrace();
            } catch (InterruptedException e1) {
                // TODO Auto-generated catch block
                e1.printStackTrace();
            }
        } else {
            System.out.println(lockInfo.lockOwner + ":上锁成功");
        }

    }

    @Override
    public void unlock() {
        try {
            if (zooKeeper.exists(lockInfo.lockname, false) != null) {
                String existOwner = new String(zooKeeper.getData(lockInfo.lockname, null, null), "UTF-8");
                if (lockInfo.lockOwner.equals(existOwner)) {
                    zooKeeper.delete(lockInfo.lockname, -1);
                    System.out.println(lockInfo.lockOwner + ": 解锁成功");
                } else {
                    System.out.println(lockInfo.lockOwner + ":无法释放锁，因为没有获得锁");
                }
            }
        } catch (UnsupportedEncodingException e) {
            e.printStackTrace();
        } catch (KeeperException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

}

2.4 测试类

package com.zhb.zookeeper.lock;

import java.io.IOException;

import org.apache.zookeeper.KeeperException;

public class LockTest {
    public static void main(String[] args) throws IOException, InterruptedException {
        LockInfo lockInfo1 = new LockInfo("/zk-lock", "person1");
        LockInfo lockInfo2 = new LockInfo("/zk-lock", "person2");
        LockInfo lockInfo3 = new LockInfo("/zk-lock", "person3");
        ZkLock zLock1 = new ZkLock(lockInfo1);
        ZkLock zLock2 = new ZkLock(lockInfo2);
        ZkLock zLock3 = new ZkLock(lockInfo3);

        new Thread(new Runnable() {

            @Override
            public void run() {
                zLock1.lock();
            }
        }).start();
        new Thread(new Runnable() {

            @Override
            public void run() {
                zLock2.lock();
            }
        }).start();

        new Thread(new Runnable() {

            @Override
            public void run() {
                zLock3.lock();
            }
        }).start();

        //停3秒钟
        Thread.sleep(3000);
        
        new Thread(new Runnable() {

            @Override
            public void run() {
                zLock1.unlock();
            }
        }).start();

        new Thread(new Runnable() {

            @Override
            public void run() {
                zLock2.unlock();
            }
        }).start();

        new Thread(new Runnable() {

            @Override
            public void run() {
                zLock3.unlock();
            }
        }).start();
        
    }
}

2.5运行结果

首先开启zookeeper服务：

然后运行测试类，结果如下：

使用zookeeper客户端查看是否正确：

正确

四、总结

本文只是zookeeper分布式锁的基础版，他的缺点是：
1.排它锁的粒度大，没有区分读、写操作，如果读多写少，则十分影响性能
2.羊群效应：锁释放后会通知所有等待中的 ZooKeeper 客户端，然后同时发起加锁请求，瞬时压力很大。

解决方案是：
1.缩小通知范围：等待锁的小伙伴们按先来后到的顺序排队吧，排好队了，接下来我只需要关心我前面一个节点的状态，当前一个节点被释放，我再去抢锁。

2.缩小锁的粒度：锁不关心业务，但是可以简单地通过操作的读、写性质来二分锁的粒度：
读锁：又称共享锁，如果前面没有写节点，可以直接上锁；当前面有写节点时，则等待距离自己最近的写节点释放（删除）。
写锁：如果前面没有节点，可以直接上锁；如果前面有节点，则等待前一个节点释放（删除）。

关注公众号： “Java不睡觉”，回复：“资源”。获取大数据全套视频和大量Java书籍

参考文章：https://mp.weixin.qq.com/s/tTpxSeEtiwZoFtU3-LOGEg