zookeeper实现分布式锁（三）

た入场券 2024-04-17 06:03 42阅读 0赞

前面我们介绍了使用zookeeper实现分布式锁的最常见的做法，就是通过创建临时节点，然后最小节点获取锁的方式。但是zookeeper实现分布式锁的方式不是这一种唯一的办法，我们知道，分布式锁的特点是同一时刻，只有一个进程能够获取锁，其他的进程获取失败，处于等待锁的状态，当获取锁的进程使用完毕之后，会释放锁资源，其他进程会再次争取这把锁。按照这个思路，再结合zookeeper的特点，如果我们同时创建一个永久节点，那么也只有一个进程会创建成功，其他的进程均失败，当我们删除这个永久节点的时候，其他进程才有可能会有一个进程创建成功。基于这个特点，我们可以利用zookeeper来创建另一种分布式锁。

通过zookeeper创建永久节点的方式创建分布式锁，和普通的锁类似，就是进程获取锁没有一个顺序，谁抢到锁了，谁就占有它，直到完成指定的任务，然后释放锁，再由其他的进程竞争这把锁。这种分布式锁更容易理解，也更加的直观。当有锁资源释放的时候，其余没有获取锁的进程同时来争抢这把锁。

获取锁的进程，也就是成功创建永久节点的进程，当释放锁的时候，会删除这个永久节点，那么其他进程就有可能会成功创建节点，这里释放锁的过程也很重要，需要明确指出永久节点的路径，才能够释放成功，即使连接不断开也没有关系，如下所示：

![20190826111330399.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][]

这个和最小临时节点获取锁的方法不同的地方在于，是所有没有获取到锁的进程都会去争这把锁，因为大家监听的节点都是一样的，没有什么先后之分，也没有优先级。这里的代码和原生的zookeeper api有些不一样，因为我们使用的是com.101tec这个依赖，下面给出完整的示例：

pom.xml文件中增加com.101tec这个依赖。

<dependency>
      <groupId>com.101tec</groupId>
      <artifactId>zkclient</artifactId>
      <version>0.10</version>
    </dependency>

定义一个锁的接口Lock.java：

package com.xxx.lock;
    public interface Lock {
    	public void lock();
    	public void unlock();
    }

定义一个抽象类，实现锁接口，ZookeeperAbstractLock.java：

package com.xxx.lock;
    import org.I0Itec.zkclient.ZkClient;
    public abstract class ZookeeperAbstractLock implements Lock {
    	private static final String CONNECTSTRING="localhost:2181";
    	protected ZkClient zkclient = new ZkClient(CONNECTSTRING);
    	protected static final String PATH = "/lock";
    	
    
    	@Override
    	public void lock() {
    		if(tryLock()){
    			System.out.println(Thread.currentThread().getName()+" 获取锁资源");
    			return;
    		}else{
    			waitLock();
    			lock();
    		}
    	}
    
    	@Override
    	public void unlock() {
    		if(zkclient!=null){
    			zkclient.delete(PATH);
    			System.out.println(Thread.currentThread().getName()+" 释放锁资源");
    			zkclient.close();
    		}
    	}
    
    	abstract boolean  tryLock();
    	abstract void waitLock();
    }

定义一个类继承抽象锁类，ZookeeperDistributedLock.java：

package com.xxx.lock;
    import java.util.concurrent.CountDownLatch;
    import org.I0Itec.zkclient.IZkDataListener;
    public class ZookeeperDistributedLock extends ZookeeperAbstractLock {
    
    	private CountDownLatch latch = null;
    	@Override
    	boolean tryLock() {
    		try {
    			zkclient.createPersistent(PATH);
    			return true;
    		} catch (Exception e) {
    			return false;
    		}
    	}
    
    	@Override
    	void waitLock() {
    		IZkDataListener listener = new IZkDataListener() {	
    			@Override
    			public void handleDataDeleted(String s) throws Exception {
    				if(latch!=null){
    					latch.countDown();
    				}
    			}
    			@Override
    			public void handleDataChange(String s, Object o) throws Exception {}
    		};
    		zkclient.subscribeDataChanges(PATH, listener);
    		if(zkclient.exists(PATH)){
    			latch = new CountDownLatch(1);
    			try {
    				latch.await();
    			} catch (Exception e) {
    				e.printStackTrace();
    			}
    		}
    		zkclient.unsubscribeDataChanges(PATH, listener);
    	}
    
    }

测试类：

package com.xxx.lock;
    public class MainService{
    
    	public static void main(String[] args) {
    		for(int i=0;i<10;i++){
    			Thread thread = new Thread(new Runnable(){
    				private Lock lock = new ZookeeperDistributedLock();			
    				@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) {
    						e.printStackTrace();
    					}finally{
    						lock.unlock();
    					}
    				}
    			});
    			thread.start();
    		}
    	}
    
    }

运行测试类，打印信息如下：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZlaW5pZmk_size_16_color_FFFFFF_t_70 2][]

因为这种锁的争锁过程是随机的，哪个进程获取锁，是无法预测的，但是获取锁之后，其余线程处于等待状态，只有获取锁的进程执行完毕，释放锁资源了，其余进程才有可能获取锁。

这种锁的实现方式和redis实现的分布式锁类似，即同一时刻，只有一个进程会成功创建永久节点，就获得锁，其余进程则等待资源释放，释放会触发监听事件，剩余进程再次争锁。

[20190826111330399.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/16/39c18ec8d2a642eabffcca64009e4ff7.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZlaW5pZmk_size_16_color_FFFFFF_t_70]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/16/57595c150b5a455886849335a5c30b6a.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/13113f1e30f548f0a6035481f792f34b.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/30ce1bbc7f844540ab5104d87546b486.png