zookeeper实现分布式锁详解

亦凉 2024-04-17 05:48 29阅读 0赞

分布式锁的实现方式有很多，之前介绍过用redis实现分布式锁，这里zookeeper也是一种常见的实现方式。在网上看了很多zookeeper实现分布式锁的例子，几乎都是介绍一个思路，然后附上一段代码，可能都没有运行过。有的是代码实现没有问题，但是示例没有体现出分布式锁的使用结果。

所谓分布式锁，其实是相对于普通的锁而言的，只不过普通的锁只能是一个进程之内的线程之间共享，当我们的程序部署在不同的机器上的时候，组成了一个分布式系统，这时候，普通的锁就不能达到我们的要求，那么锁资源需要是所有的进程都能够访问，并且只能有一个进程的某一个线程能够获取这个资源，这时候的锁资源，就需要是一个服务，比如mysql，redis，zookeeper甚至只要是能够通过ip和端口能够访问的服务，都可以作为这个锁。

redis作为分布式锁的实现是使用setnx(key,value)这个操作，就是同一时间，只能有一个进程能够设置成功，其他的进程只能等待这个进程释放锁，也就是delete这个key。

同样的，如果是mysql，那么可以利用mysql中建表特点，同一时间也是只能有一个进程的建表语句能够建立同一个名称的表，其他的进程会失败，需要等待该进程执行完他的操作然后释放锁，即删除这个表之后，才能继续获取锁，即创建表。

zookeeper实现分布式锁，也是一样的道理，我们会在同一个路径下创建一批临时节点，这些临时节点是有序列号的，每次只让序列号最小的节点获取锁，其他的节点等待，并且监听比他小的节点删除事件，也是等待锁的过程，当最小的节点获取锁资源，并执行完相关操作，他会释放锁，也就是删除这个临时节点。这时候，注册了这个路径的监听事件的节点会获取锁，还是一样的思路，序列号最小的节点获取锁，依次类推，直到所有的节点均获取锁并执行相关操作。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZlaW5pZmk_size_16_color_FFFFFF_t_70][]

zookeeper实现分布式锁的原理这里就介绍完了，zookeeper分布式锁的特点在于，创建的临时节点是唯一的，而且序列号是按照顺序递增的，序列号可以看做是一个长整型数据。

现在来看通过程序如何实现这个分布式锁，有几个需要注意的地方：

1、默认我们需要在一个路径，比如/root下创建临时节点，而且创建临时节点的路径，我们还需要指定一个容易区分的名称，比如/root/\_locknode\_，通过api，我们可以如下方式来创建这个临时节点。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZlaW5pZmk_size_16_color_FFFFFF_t_70 1][]

2、正常获取锁的逻辑判断很简单，就是通过/root这个路径找到它的所有子节点，然后找到序列号最小的节点即可。

3、如果不是最小节点，那么就需要另行处理了，这里需要等待最小节点释放锁，而按照之前的介绍，下一个获取锁的节点会是最小节点删除之后的节点当中的最小节点，也就是第二小的节点。所以在等待获取锁的过程中，我们需要监听比当前节点还要小的一个节点，而不是所有比当前节点都小的节点。因此，监听节点变化也是一个按照顺序依次监听的过程，\_locknode\_0000000009监听节点\_locknode\_0000000008，\_locknode\_0000000008监听\_locknode\_0000000007，依此类推，\_locknode\_0000000001监听\_locknode\_0000000000，当前一个被监听节点删除的时候，当前节点就获取锁。

4、这里释放锁的过程就显得很重要了，如果释放的不正确，后续的节点就无法按照这个获取锁的逻辑来获取锁了。

5、每一个节点最终释放锁的时候，需要删除该节点，因此我们需要在锁的实体类中定义一个私有变量，记录这个锁的路径，这里以myName为例。

6、当一个节点释放锁资源之后，其他节点监听到变化了，这时候就需要结束等待，并且需要再次获取锁，一般来说，就是需要去抢一次锁资源，就是需要递归调用加锁方法。

有了这个思路，我们的分布式锁就可以实现了，这里给出一个示例：

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

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

实现这个锁，DistributedLock.java

package com.xxx.lock3;
    import java.util.Collections;
    import java.util.List;
    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 DistributedLock implements Lock{
    	private String path = "/root/_locknode_";
    	private ZooKeeper zk;
    	private static final int sessionTimeout = 6000;
    	private CountDownLatch latch = new CountDownLatch(1);
    	private String myName;
    	
    	public DistributedLock(){
    		try {
    			zk = new ZooKeeper("localhost:2181", sessionTimeout, new Watcher() {	
    				@Override
    				public void process(WatchedEvent event) {
    					if(event.getState()==KeeperState.SyncConnected){
    						latch.countDown();
    					}
    				}
    			});
    			latch.await();
    			myName = zk.create(path, null, Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
    		} catch (Exception e) {
    			e.printStackTrace();
    		}
    	}
    
    	@Override
    	public void lock() {
    		try {
    			if(tryLock()){
    				System.out.println(Thread.currentThread().getName()+" get lock.");
    				return;
    			}else{
    				waitLock();
    				lock();
    			}
    		} catch (KeeperException e) {
    			// TODO Auto-generated catch block
    			e.printStackTrace();
    		} catch (InterruptedException e) {
    			// TODO Auto-generated catch block
    			e.printStackTrace();
    		}
    	}
    
    	@Override
    	public void unlock() {
    		try {
    			zk.delete(myName, -1);
    			zk.close();
    		} catch (Exception e) {
    			// TODO: handle exception
    		}
    	}
    	
    	public boolean tryLock() throws KeeperException, InterruptedException{
    		List<String> list = zk.getChildren("/root", false);
    		String[] mystr = myName.split("_");
    		
    		long myId = Long.parseLong(mystr[2]);
    		boolean minId = true;
    		for(String childName:list){
    			String[] childNames = childName.split("_");
    			long id = Long.parseLong(childNames[2]);
    			if(id<myId){
    				minId = false;
    				break;
    			}
    		}
    		if(minId)
    			return true;
    		return false;
    	}
    	
    	public void waitLock(){
    		try {
    			CountDownLatch latch = new CountDownLatch(1);
    			List<String> children = zk.getChildren("/root", false);
    			Collections.sort(children);
    			String myId = myName.split("/")[2];
    			int index = children.indexOf(myId);
    			if(index<=0){
    				throw new IllegalArgumentException("data error");
    			}
    			//String headPath = children.get(index-1);
    			String headPath = children.get(0);
    			zk.exists("/root/"+headPath, new Watcher(){
    				@Override
    				public void process(WatchedEvent event) {
    					if(event.getType()==EventType.NodeDeleted){
    						latch.countDown();
    					}
    				}	
    			});
    			latch.await();
    		} catch (Exception e) {
    			// TODO Auto-generated catch block
    			e.printStackTrace();
    		}
    	}
    	
    }

测试程序，启动十个线程，每个线程执行同样的run方法，通过使用分布式锁，我们的代码如下：

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

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

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZlaW5pZmk_size_16_color_FFFFFF_t_70 2][]

这个示例，虽然是在单机上运行的，但是也能看出来加锁，释放锁期间，只有一个线程在运行，其他的线程均在等待锁的释放，这里的结果也验证了，通过zookeeper实现的分布式锁，加锁释放锁的过程是一个类似塔罗牌的过程，锁的获取顺序是可以预见的。这个代码有一个小问题，就是我这里是默认/root路径时存在的，如果你运行程序发现问题，可以在zookeeper上手工创建/root节点。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZlaW5pZmk_size_16_color_FFFFFF_t_70]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/11/18d35fda1a8b4da3aed2b6de97c5bb03.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZlaW5pZmk_size_16_color_FFFFFF_t_70 1]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/11/7708aeeddba644179ebf71fdc5d6811d.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZlaW5pZmk_size_16_color_FFFFFF_t_70 2]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/11/0cb2c4bd7cbe4147a2837e01857d5b40.png