zookeeper实现分布式锁（二）

深藏阁楼爱情的钟 2024-04-17 06:03 59阅读 0赞

前面通过示例详细介绍了通过zookeeper实现分布式锁的原理和过程，我们了解到，zookeeper实现分布式锁的过程类似塔罗牌，就是zookeeper会在一个路径下创建一批临时节点，并且根据序列号最小的节点获取锁，锁用完之后，会释放，即删除这个临时节点，同时会触发监听了该节点的节点停止等待，重新获取锁。

如果一个永久节点比如：/root下创建了10个临时节点，分别为：/root/\_locknode\_0000000000，

/root/\_locknode\_0000000001,.....,/root/\_locknode\_0000000009。按照最小序列号节点获取锁的机制，可以预见这10个节点组成的集合，会依次按照节点从小到大的顺序获取锁，并释放锁。

一般来说，当一个节点获取锁之后，其他的节点会等待锁，当该节点锁使用完毕，释放锁的时候，会触发zookeeper删除节点NodeDelete事件，这个事件可以通知其他节点结束等待，然后重新获取锁，就是一个递归调用锁的过程。但是这里的删除节点事件，不用通知其他的每个节点去争锁，因为按照zookeeper分布式锁的原理，只有监听了当前节点，并且序列号最小的节点才会获取锁。我们可以有效控制监听的范围，减少不必要的争锁。就是说等待的节点都会监听一个节点，但是不是最小的节点，而是比它小的最大的节点，似乎有点绕口，就是比他小的相邻节点。以上面的节点为例，我们可以这样控制，让0000000001监听0000000000,让0000000002监听0000000001，...，依此类推，让0000000009监听0000000008。这样的话，我们控制了监听的范围，当0000000000节点删除的时候，不用每一个等待的节点都去争锁，只让0000000001去获得锁即可，其实都不用去获得，因为已经轮到它了，所以在编码中，结束等待，直接return就可以了，也不用递归调用获取锁的方法，等待锁的代码如下所示：

![20190825234026166.png][]

这里等待结束，就获得锁。因为上一个节点释放锁的时候，会删除节点，触发删除节点事件：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZlaW5pZmk_size_16_color_FFFFFF_t_70][]

通知事件：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZlaW5pZmk_size_16_color_FFFFFF_t_70 1][]

这里的删除节点通知事件会触发监听节点停止等待操作，停止的时候，其实就是监听节点获取锁的时刻（不用调用争锁的方法，因为轮到它获取锁），这里使用了一个很巧妙的办法。

这里给出完整的代码：

package com.xxx.lock2;
    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.WatchedEvent;
    import org.apache.zookeeper.Watcher;
    import org.apache.zookeeper.Watcher.Event.KeeperState;
    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="/root";
    	private String lockName = "_locknode_";
        private String waitNode;
        private String curNode;
        private CountDownLatch latch;
        private CountDownLatch connectedSignal = new CountDownLatch(1);
        
        public DistributedLock(String config){
        	try {
    			zk = new ZooKeeper(config, 6000, this);
    			connectedSignal.await();
    			Stat stat = zk.exists(root, false);
    			if(stat==null){
    				zk.create(root, new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
    			}
    		} catch (Exception e) {
    			// TODO: handle exception
    		}
        }
    	
    	public static void main(String[] args) throws InterruptedException {
    		for(int i=0;i<10;i++){
    			Thread thread = new Thread(new Runnable() {
    				DistributedLock lock = new DistributedLock("localhost:2181");
    				@Override
    				public void run() {
    					try {
    						lock.lock();
    						System.out.println(Thread.currentThread().getName()+" start.");
    						Thread.sleep(200);
    						System.out.println(Thread.currentThread().getName()+ " end.");
    					} catch (Exception e) {
    						// TODO: handle exception
    					}finally{
    						lock.unlock();
    					}
    				}
    			});
    			thread.start();
    		}
    	}
    
    	@Override
    	public void process(WatchedEvent event) {
    		// TODO Auto-generated method stub
    		if(event.getState()==KeeperState.SyncConnected){
    			connectedSignal.countDown();
    		}
    		if(this.latch!=null){
    			this.latch.countDown();
    		}
    	}
    
    	@Override
    	public void lock() {
    		// TODO Auto-generated method stub
    		try {
    			if(this.tryLock()){
    				System.out.println(Thread.currentThread().getName()+" get lock.");
    				return;
    			}else{
    				waitForNode(waitNode);
    			}
    		} catch (Exception e) {
    			// TODO: handle exception
    			throw new RuntimeException(e);
    		}
    	}
    
    	@Override
    	public void lockInterruptibly() throws InterruptedException {
    		// TODO Auto-generated method stub
    		this.lock();
    	}
    
    	@Override
    	public boolean tryLock() {
    		// TODO Auto-generated method stub
    		try {
    			curNode = zk.create(root+"/"+lockName, new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
    			List<String> children = zk.getChildren(root, false);
    			Collections.sort(children);
    			if(curNode.equals(root+"/"+children.get(0))){
    				return true;
    			}
    			String subNode = curNode.substring(curNode.lastIndexOf("/")+1);
    			waitNode = children.get(Collections.binarySearch(children, subNode)-1);
    		} catch (Exception e) {
    			// TODO: handle exception
    		}
    		return false;
    	}
    
    	@Override
    	public boolean tryLock(long time, TimeUnit unit)
    			throws InterruptedException {
    		// TODO Auto-generated method stub
    		try {
    			if(this.tryLock()){
    				return true;
    			}
    			return waitForNode(waitNode);	
    		} catch (Exception e) {
    			// TODO: handle exception
    		}
    		return false;
    	}
    
    	@Override
    	public void unlock() {
    		// TODO Auto-generated method stub
    		try {
    			zk.delete(curNode, -1);
    			zk.close();
    		} catch (Exception e) {
    			// TODO Auto-generated catch block
    			e.printStackTrace();
    		}
    	}
    
    	@Override
    	public Condition newCondition() {
    		// TODO Auto-generated method stub
    		return null;
    	}
    
    	public boolean waitForNode(String path) throws Exception{
    		Stat stat = zk.exists(root+"/"+path, true);
    		if(stat!=null){
    			this.latch = new CountDownLatch(1);
    			this.latch.await();
    			this.latch = null;
    		}
    		return true;
    	}
    }

运行程序，打印结果如下：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZlaW5pZmk_size_16_color_FFFFFF_t_70 2][]

从打印结果来看，结果符合分布式锁的预期，塔罗牌的方式获取锁，而且只有第一个获取锁的过程是真正的在计算最小节点，其他的节点在第一个节点获取锁的时候，处在等待锁的状态，但是已经能够预见获取锁的顺序。因此不用递归调用争锁的方法，只需要监听比当前节点序列号小的相邻节点删除事件，等待结束，就获取锁。

[20190825234026166.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/16/659f34d209124129baf650b82c2af0a7.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZlaW5pZmk_size_16_color_FFFFFF_t_70]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/16/e802cd0c33ca4218be8759b3053a6f2f.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZlaW5pZmk_size_16_color_FFFFFF_t_70 1]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/16/70bf139562154979ab7206c1b69e32d1.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZlaW5pZmk_size_16_color_FFFFFF_t_70 2]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/16/728c27c79db14b978002d736a4090c69.png