Zookeeeper应用实践（四）

zk的应用还是非常广泛的。

1. 分布式锁

单机环境下的锁还是很容易去实现的，但是在分布式环境下一切都变得不是那么简单。zk实现分布式锁的原理还简单，因为在分布式环境中的zk节点的变化会被每一台机器watch，有任何变化都会被通知，所以我们可以利用这个机制：

1. 我们创建一个根节点Lock，在获取锁的时候在该节点下创建临时顺序节点，释放锁的时候删除该临时节点。客户端调用createNode方法在Lock下创建临时顺序节点。
2. 调用getChildren(” Lock”)来获取Lock下的所有节点，注意此时不用设置任何的Watcher。
3. 客户端获取到所有子节点的path之后发现如果自己之前所创建的子节点序号最小，那么就认为该客户端获取到了锁。如果发现自己所创的节点并非Lock中最小的说明自己还没获得到锁。
4. 此时客户端需要找到比自己小的马哥节点，然后对其调用exist()方法，同时对其注册时间监听器。是山猫关注这个节点是否被删除（前面我们说了释放所就会删除节点）。
5. 当监听到了删除事件，再次判断自己所创建的节点是否是Lock中节点顺序最小的节点，如果是则获取到了锁，如果不是则继续重复以上步骤直到获取到比自己小的节点的删除事件并注册监听。

大致流程就是上面所说，我们可以画一个流程图来表示：

下面写一个测试例子运行一下我们的逻辑：

DistributedLock：

import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;

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;


public class DistributedLock implements Lock, Watcher{
    private ZooKeeper zk;
    private String root = "/locks";//根
    private String lockName;//竞争资源的标志
    private String waitNode;//等待前一个锁
    private String myZnode;//当前锁
    private CountDownLatch latch;//计数器
    private int sessionTimeout = 30000;
    private List<Exception> exception = new ArrayList<Exception>();

    /**
     * 创建分布式锁,使用前请确认config配置的zookeeper服务可用
     * @param config 127.0.0.1:2181
     * @param lockName 竞争资源标志,lockName中不能包含单词lock
     */
    public DistributedLock(String config, String lockName){
        this.lockName = lockName;
        // 创建一个与服务器的连接
        try {
            zk = new ZooKeeper(config, sessionTimeout, this);
            Stat stat = zk.exists(root, false);
            if(stat == null){
                // 创建根节点
                zk.create(root, new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE,CreateMode.PERSISTENT);
            }
        } catch (IOException e) {
            exception.add(e);
        } catch (KeeperException e) {
            exception.add(e);
        } catch (InterruptedException e) {
            exception.add(e);
        }
    }

    /**
     * zookeeper节点的监视器
     */
    public void process(WatchedEvent event) {
        if(this.latch != null) {
            this.latch.countDown();
        }
    }

    public void lock() {
        if(exception.size() > 0){
            throw new LockException(exception.get(0));
        }
        try {
            if(this.tryLock()){
                System.out.println("Thread " + Thread.currentThread().getId() + " " +myZnode + " get lock true");
                return;
            }
            else{
                waitForLock(waitNode, sessionTimeout);//等待锁
            }
        } catch (KeeperException e) {
            throw new LockException(e);
        } catch (InterruptedException e) {
            throw new LockException(e);
        }
    }

    public boolean tryLock() {
        try {
            String splitStr = "_lock_";
            if(lockName.contains(splitStr))
                throw new LockException("lockName can not contains \u000B");
            //创建临时子节点
            myZnode = zk.create(root + "/" + lockName + splitStr, new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE,CreateMode.EPHEMERAL_SEQUENTIAL);
            System.out.println(myZnode + " is created ");
            //取出所有子节点
            List<String> subNodes = zk.getChildren(root, false);
            //取出所有lockName的锁
            List<String> lockObjNodes = new ArrayList<String>();
            for (String node : subNodes) {
                String _node = node.split(splitStr)[0];
                if(_node.equals(lockName)){
                    lockObjNodes.add(node);
                }
            }
            Collections.sort(lockObjNodes);
            System.out.println(myZnode + "==" + lockObjNodes.get(0));
            if(myZnode.equals(root+"/"+lockObjNodes.get(0))){
                //如果是最小的节点,则表示取得锁
                return true;
            }
            //如果不是最小的节点，找到比自己小1的节点
            String subMyZnode = myZnode.substring(myZnode.lastIndexOf("/") + 1);
            waitNode = lockObjNodes.get(Collections.binarySearch(lockObjNodes, subMyZnode) - 1);
        } catch (KeeperException e) {
            throw new LockException(e);
        } catch (InterruptedException e) {
            throw new LockException(e);
        }
        return false;
    }

    public boolean tryLock(long time, TimeUnit unit) {
        try {
            if(this.tryLock()){
                return true;
            }
            return waitForLock(waitNode,time);
        } catch (Exception e) {
            e.printStackTrace();
        }
        return false;
    }

    private boolean waitForLock(String lower, long waitTime) throws InterruptedException, KeeperException {
        Stat stat = zk.exists(root + "/" + lower,true);
        //判断比自己小一个数的节点是否存在,如果不存在则无需等待锁,同时注册监听
        if(stat != null){
            System.out.println("Thread " + Thread.currentThread().getId() + " waiting for " + root + "/" + lower);
            this.latch = new CountDownLatch(1);
            this.latch.await(waitTime, TimeUnit.MILLISECONDS);
            this.latch = null;
        }
        return true;
    }

    public void unlock() {
        try {
            System.out.println("unlock " + myZnode);
            zk.delete(myZnode,-1);
            myZnode = null;
            zk.close();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (KeeperException e) {
            e.printStackTrace();
        }
    }

    public void lockInterruptibly() throws InterruptedException {
        this.lock();
    }

    public Condition newCondition() {
        return null;
    }

    public class LockException extends RuntimeException {
        private static final long serialVersionUID = 1L;
        public LockException(String e){
            super(e);
        }
        public LockException(Exception e){
            super(e);
        }
    }

}

这里面的实现逻辑和上面叙述的一样。然后我们创建一个多线程环境的测试：

import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicInteger;



public class ConcurrentTest {
    private CountDownLatch startSignal = new CountDownLatch(1);//开始阀门
    private CountDownLatch doneSignal = null;//结束阀门
    private CopyOnWriteArrayList<Long> list = new CopyOnWriteArrayList<Long>();
    private AtomicInteger err = new AtomicInteger();//原子递增
    private ConcurrentTask[] task = null;

    public ConcurrentTest(ConcurrentTask... task){
        this.task = task;
        if(task == null){
            System.out.println("task can not null");
            System.exit(1);
        }
        doneSignal = new CountDownLatch(task.length);
        start();
    }
    /**
     * @param args
     * @throws ClassNotFoundException
     */
    private void start(){
        //创建线程，并将所有线程等待在阀门处
        createThread();
        //打开阀门
        startSignal.countDown();//递减锁存器的计数，如果计数到达零，则释放所有等待的线程
        try {
            doneSignal.await();//等待所有线程都执行完毕
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        //计算执行时间
        getExeTime();
    }
    /**
     * 初始化所有线程，并在阀门处等待
     */
    private void createThread() {
        long len = doneSignal.getCount();
        for (int i = 0; i < len; i++) {
            final int j = i;
            new Thread(new Runnable(){
                public void run() {
                    try {
                        startSignal.await();//使当前线程在锁存器倒计数至零之前一直等待
                        long start = System.currentTimeMillis();
                        task[j].run();
                        long end = (System.currentTimeMillis() - start);
                        list.add(end);
                    } catch (Exception e) {
                        err.getAndIncrement();//相当于err++
                    }
                    doneSignal.countDown();
                }
            }).start();
        }
    }
    /**
     * 计算平均响应时间
     */
    private void getExeTime() {
        int size = list.size();
        List<Long> _list = new ArrayList<Long>(size);
        _list.addAll(list);
        Collections.sort(_list);
        long min = _list.get(0);
        long max = _list.get(size-1);
        long sum = 0L;
        for (Long t : _list) {
            sum += t;
        }
        long avg = sum/size;
        System.out.println("min: " + min);
        System.out.println("max: " + max);
        System.out.println("avg: " + avg);
        System.out.println("err: " + err.get());
    }

    public interface ConcurrentTask {
        void run();
    }

}

接下来写一个测试方法：

public class ZkTest {
    public static void main(String[] args) {
        Runnable task1 = new Runnable(){
            public void run() {
                DistributedLock lock = null;
                try {
                    lock = new DistributedLock("192.168.131.128:2181","test1");
                    //lock = new DistributedLock("127.0.0.1:2182","test2");
                    lock.lock();
                    Thread.sleep(3000);
                    System.out.println("===Thread " + Thread.currentThread().getId() + " running");
                } catch (Exception e) {
                    e.printStackTrace();
                }
                finally {
                    if(lock != null)
                        lock.unlock();
                }

            }

        };
        new Thread(task1).start();
        try {
            Thread.sleep(1000);
        } catch (InterruptedException e1) {
            e1.printStackTrace();
        }
        ConcurrentTest.ConcurrentTask[] tasks = new ConcurrentTest.ConcurrentTask[10];
        for(int i=0;i<tasks.length;i++){
            ConcurrentTest.ConcurrentTask task3 = new ConcurrentTest.ConcurrentTask(){
                public void run() {
                    DistributedLock lock = null;
                    try {
                        lock = new DistributedLock("192.168.131.130:2181","test2");
                        lock.lock();
                        System.out.println("Thread " + Thread.currentThread().getId() + " running");
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                    finally {
                        lock.unlock();
                    }

                }
            };
            tasks[i] = task3;
        }
        new ConcurrentTest(tasks);
    }
}

运行结果如下：

/locks/test1_lock_0000000001 is created 
/locks/test1_lock_0000000001==test1_lock_0000000001
Thread 12 /locks/test1_lock_0000000001 get lock true
/locks/test2_lock_0000000002 is created 
/locks/test2_lock_0000000003 is created 
/locks/test2_lock_0000000004 is created 
/locks/test2_lock_0000000005 is created 
/locks/test2_lock_0000000002==test2_lock_0000000002
Thread 18 /locks/test2_lock_0000000002 get lock true
Thread 18 running
unlock /locks/test2_lock_0000000002
/locks/test2_lock_0000000003==test2_lock_0000000002
/locks/test2_lock_0000000004==test2_lock_0000000002
/locks/test2_lock_0000000005==test2_lock_0000000002
/locks/test2_lock_0000000006 is created 
Thread 24 waiting for /locks/test2_lock_0000000003
Thread 21 running
unlock /locks/test2_lock_0000000003
Thread 22 waiting for /locks/test2_lock_0000000004
/locks/test2_lock_0000000007 is created 
/locks/test2_lock_0000000008 is created 
/locks/test2_lock_0000000009 is created 
/locks/test2_lock_0000000006==test2_lock_0000000003
/locks/test2_lock_0000000010 is created 
/locks/test2_lock_0000000011 is created 
/locks/test2_lock_0000000008==test2_lock_0000000004
Thread 24 running
unlock /locks/test2_lock_0000000004
/locks/test2_lock_0000000007==test2_lock_0000000004
/locks/test2_lock_0000000009==test2_lock_0000000004
Thread 20 waiting for /locks/test2_lock_0000000005
/locks/test2_lock_0000000011==test2_lock_0000000004
/locks/test2_lock_0000000010==test2_lock_0000000004
Thread 15 waiting for /locks/test2_lock_0000000007
Thread 23 waiting for /locks/test2_lock_0000000006
Thread 22 running
unlock /locks/test2_lock_0000000005
Thread 19 waiting for /locks/test2_lock_0000000008
Thread 17 waiting for /locks/test2_lock_0000000010
Thread 16 waiting for /locks/test2_lock_0000000009
Thread 20 running
unlock /locks/test2_lock_0000000006
Thread 23 running
unlock /locks/test2_lock_0000000007
Thread 15 running
unlock /locks/test2_lock_0000000008
Thread 19 running
unlock /locks/test2_lock_0000000009
Thread 16 running
unlock /locks/test2_lock_0000000010
Thread 17 running
unlock /locks/test2_lock_0000000011
min: 93
max: 156
avg: 124
err: 0
===Thread 12 running
unlock /locks/test1_lock_0000000001

Process finished with exit code 0

通过测试例子中：

第一次的测试只创建了一个节点test1_01,很显然是可以成功获取到锁的。

下面的测试中创建了10组线程，可以发现他们之间是有锁竞争的。到最后每个线程严格按照节点创建顺序分别都获取到了锁。