七、zookeeper使用场景之：分布式锁

一、锁的的基本概念

开发中锁的概念并不陌生，通过锁可以实现在多个线程或多个进程间在争抢资源时，能够合理的分配置资源的所有权。在单体应用中我们可以通过 synchronized 或ReentrantLock 来实现锁。但在分布式系统中，仅仅是加synchronized 是不够的，需要借助第三组件来实现。比如一些简单的做法是使用关系型数据行级锁来实现不同进程之间的互斥，但大型分布式系统的性能瓶颈往往集中在数据库操作上。为了提高性能得采用如Redis、Zookeeper之内的组件实现分布式锁。

二、锁的获取

某银行帐户，可以同时进行帐户信息的读取，但读取其间不能修改帐户数据。其帐户ID为:888

获得读锁流程：

1、基于资源ID创建临时序号读锁节点 /lock/888.R0000000002 Read
2、获取 /lock 下所有子节点，判断其最小的节点是否为读锁，如果是则获锁成功
3、最小节点不是读锁，则阻塞等待。添加lock/ 子节点变更监听。
4、当节点变更监听触发，执行第2步

数据结构：

获得写锁流程：
1、基于资源ID创建临时序号写锁节点 /lock/888.R0000000002 Write
2、获取 /lock 下所有子节点，判断其最小的节点是否为自己，如果是则获锁成功
3、最小节点不是自己，则阻塞等待。添加lock/ 子节点变更监听。
4、当节点变更监听触发，执行第2步
释放锁：
读取完毕后，手动删除临时节点，如果获锁期间宕机，则会在会话失效后自动删除。

三、关于羊群效应

在等待锁获得期间，所有等待节点都在监听 Lock节点，一但lock 节点变更所有等待节点都会被触发，然后在同时反查Lock 子节点。如果等待对例过大会使用Zookeeper承受非常大的流量压力。

为了改善这种情况，可以采用监听链表的方式，每个等待对列只监听前一个节点，如果前一个节点释放锁的时候，才会被触发通知。这样就形成了一个监听链表。

四、代码实例

package com.suibibk.zookeeper;
public class Lock {
    private String lockId;
    private String path;
    private boolean active;
    public Lock(String lockId, String path) {
        this.lockId = lockId;
        this.path = path;
    }
    public Lock() {
    }
    public String getLockId() {
        return lockId;
    }
    public void setLockId(String lockId) {
        this.lockId = lockId;
    }
    public String getPath() {
        return path;
    }
    public void setPath(String path) {
        this.path = path;
    }
    public boolean isActive() {
        return active;
    }
    public void setActive(boolean active) {
        this.active = active;
    }
}

public class ZookeeperLock {
    private String server = "192.168.209.4:2181";
    private ZkClient zkClient;
    private static final String rootPath = "/tuling-lock";
    public ZookeeperLock() {
        zkClient = new ZkClient(server, 5000, 20000);
        buildRoot();
    }
    // 构建根节点
    public void buildRoot() {
        if (!zkClient.exists(rootPath)) {
            zkClient.createPersistent(rootPath);
        }
    }
    public Lock lock(String lockId, long timeout) {
        Lock lockNode = createLockNode(lockId);
        lockNode = tryActiveLock(lockNode);// 尝试激活锁
        if (!lockNode.isActive()) {
            try {
                synchronized (lockNode) {
                    lockNode.wait(timeout);
                }
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
        }
        if (!lockNode.isActive()) {
            throw new RuntimeException(" lock  timeout");
        }
        return lockNode;
    }
    public void unlock(Lock lock) {
        if (lock.isActive()) {
            zkClient.delete(lock.getPath());
        }
    }
    // 尝试激活锁
    private Lock tryActiveLock(Lock lockNode) {
        // 判断当前是否为最小节点
        List<String> list = zkClient.getChildren(rootPath)
                .stream()
                .sorted()
                .map(p -> rootPath + "/" + p)
                .collect(Collectors.toList());
        String firstNodePath = list.get(0);
        if (firstNodePath.equals(lockNode.getPath())) {
            lockNode.setActive(true);
        } else {
            String upNodePath = list.get(list.indexOf(lockNode.getPath()) - 1);
            zkClient.subscribeDataChanges(upNodePath, new IZkDataListener() {
                @Override
                public void handleDataChange(String dataPath, Object data) throws Exception {
                }
                @Override
                public void handleDataDeleted(String dataPath) throws Exception {
                    // 事件处理 与心跳 在同一个线程，如果Debug时占用太多时间，将导致本节点被删除，从而影响锁逻辑。
                    System.out.println("节点删除:" + dataPath);
                     Lock lock = tryActiveLock(lockNode);
                    synchronized (lockNode) {
                        if (lock.isActive()) {
                            lockNode.notify();
                        }
                    }
                    zkClient.unsubscribeDataChanges(upNodePath, this);
                }
            });
        }
        return lockNode;
    }
    public Lock createLockNode(String lockId) {
        String nodePath = zkClient.createEphemeralSequential(rootPath + "/" + lockId, "lock");
        return new Lock(lockId, nodePath);
    }
    static int i = 0;
    static String lockId = "lockId";
    public static void main(String[] args) {
        ZookeeperLock lock = new ZookeeperLock();
        for (int j = 0; j < 100; j++) {
            new Thread(new Runnable() {
                @Override
                public void run() {
                    try {
                        Thread.currentThread().sleep(100l);
                    } catch (InterruptedException e) {
                        // TODO Auto-generated catch block
                        e.printStackTrace();
                    }
                    Lock l = lock.createLockNode(lockId);
                    i=i+1;
                    System.out.println(Thread.currentThread()+"i="+i);
                    lock.unlock(l);
                }
            }).start();
        }
    }
}

运行上面的例子，查看日志可以看成功锁住了

Thread[Thread-71,5,main]i=1
Thread[Thread-44,5,main]i=2
Thread[Thread-20,5,main]i=3
Thread[Thread-24,5,main]i=4
Thread[Thread-7,5,main]i=5
Thread[Thread-21,5,main]i=6
...
Thread[Thread-33,5,main]i=91
Thread[Thread-49,5,main]i=92
Thread[Thread-34,5,main]i=93
Thread[Thread-58,5,main]i=94
Thread[Thread-30,5,main]i=95
Thread[Thread-61,5,main]i=96
Thread[Thread-48,5,main]i=97
Thread[Thread-29,5,main]i=98
Thread[Thread-69,5,main]i=99
Thread[Thread-62,5,main]i=100