zookeeper--zookeeper分布式锁原理

Posted 2021-02-13 flgb

目前几乎很多大型网站及应用都是分布式部署的，分布式场景中的数据一致性问题一直是一个比较重要的话题。分布式的CAP理论告诉我们“任何一个分布式系统都无法同时满足一致性（Consistency）、可用性（Availability）和分区容错性（Partition tolerance），最多只能同时满足两项。”所以，很多系统在设计之初就要对这三者做出取舍。在互联网领域的绝大多数的场景中，都需要牺牲强一致性来换取系统的高可用性，系统往往只需要保证“最终一致性”，只要这个最终时间是在用户可以接受的范围内即可。

在很多场景中，我们为了保证数据的最终一致性，需要很多的技术方案来支持，比如分布式事务、分布式锁等。有的时候，我们需要保证一个方法在同一时间内只能被同一个线程执行。在单机环境中，Java中其实提供了很多并发处理相关的API，但是这些API在分布式场景中就无能为力了。也就是说单纯的Java Api并不能提供分布式锁的能力。所以针对分布式锁的实现目前有多种方案。

针对分布式锁的实现，目前比较常用的有以下几种方案：

　　基于数据库实现分布式锁 基于缓存（redis，memcached，tair）实现分布式锁 基于Zookeeper实现分布式锁

　一、zookeeper中分布式锁实现原理

（1）、普通节点思路

现在模拟一个使用Zookeeper实现分布式锁，假设有A,B,C三台客户端去访问资源，调用zookeeper获得锁。客户端三个在zookeeper的 /locks节点下创建一个/lock节点，由于节点是唯一性的特性，只有一个人会创建成功，其余两个创建失败，会进入监听/locks节点的变化，如果/locks下子节点/lock节点发生变化，其余两个可以去拿锁，这样是否好呢？ 这样子会导致惊群效应。就是一个触发使得在短时间呢会触发大量的watcher事件，但是只有一个客户端能拿到锁

　　--众人抢，大量watcher事件

（2）、有序节点思路

1、在获取分布式锁的时候在locker节点下创建临时顺序节点，释放锁的时候删除该临时节点。

2、客户端调用createNode方法在locks下创建临时顺序节点，然后调用getChildren(“locks”)来获取locks下面的所有子节点，注意此时不用设置任何Watcher。

3、客户端获取到所有的子节点path之后，如果发现自己创建的子节点序号最小，那么就认为该客户端获取到了锁。

4、如果发现自己创建的节点并非locks所有子节点中最小的，说明自己还没有获取到锁，此时客户端需要找到比自己小的那个节点，然后对其调用exist()方法，同时对其注册事件监听器。

5、之后，让这个被关注的节点删除，则客户端的Watcher会收到相应通知，此时再次判断自己创建的节点是否是locks子节点中序号最小的，如果是则获取到了锁，如果不是则重复以上步骤继续获取到比自己小的一个节点并注册监听。

　　

二、代码实现

package com.lf.zookeeper.lock;

import java.util.List;
import java.util.SortedSet;
import java.util.TreeSet;
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 DestributeLock implements Lock,Watcher{
	
	private ZooKeeper zk = null;
	private String ROOT_LOCK ="/locks";//定义根节点
	private String CURRENT_LOCK;//当前锁
	private String WAIT_LOCK ;//等待前一个对象释放锁
	
	private CountDownLatch countDownLatch;
	
	
	public DestributeLock() {
		try {
			zk= new ZooKeeper("192.168.25.129:2181,192.168.25.130:2181,192.168.25.131:2181", 4000, this);
			//判断根节点是否存在
			Stat stat = zk.exists(ROOT_LOCK, false);
			if(stat==null){
				zk.create(ROOT_LOCK, "1".getBytes(),ZooDefs.Ids.OPEN_ACL_UNSAFE , CreateMode.PERSISTENT);
			}
		} catch (Exception e) {
			e.printStackTrace();
		}
	}

	@Override
	public void process(WatchedEvent event) {
			
		if(countDownLatch != null){
			this.countDownLatch.countDown();
		}
	}

	@Override
	public void lock() {
		if(tryLock()){
			System.out.println(Thread.currentThread().getName()+"->"+CURRENT_LOCK+",获取锁成功！");
			return;
		}
		try {
			waitForLock(WAIT_LOCK);//如果没有获得锁，继续等待
		} catch (Exception e) {
			e.printStackTrace();
		}
	}

	private boolean waitForLock(String prev) throws Exception, InterruptedException {
		Stat stat = zk.exists(prev, true);
		if(stat!=null){
			System.out.println(Thread.currentThread().getName()+"->等待"+ROOT_LOCK+prev+"释放锁");
			countDownLatch = new CountDownLatch(1);
			countDownLatch.await();
			System.out.println(Thread.currentThread().getName()+"->"+"获得锁成功！");
		}
		return true;
	}

	@Override
	public void lockInterruptibly() throws InterruptedException {
		// TODO Auto-generated method stub
		
	}

	@Override
	public Condition newCondition() {
		// TODO Auto-generated method stub
		return null;
	}

	@Override
	public boolean tryLock() {
		// TODO Auto-generated method stub
		try {
			
			CURRENT_LOCK= zk.create(ROOT_LOCK+"/", "0".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
			System.out.println(Thread.currentThread().getName()+"->"+CURRENT_LOCK+",尝试竞争锁！");
			//获取根节点下的所有子节点
			List<String> childrens = zk.getChildren(ROOT_LOCK, false);
			SortedSet<String> sortedSet = new TreeSet();//定义一个有序的集合进行排序
			for (String children : childrens) {
				sortedSet.add(ROOT_LOCK+"/"+children);
			}
			//获取最小的子节点
			String firstNode = sortedSet.first();
			SortedSet<String> lessthanMe = sortedSet.headSet(CURRENT_LOCK);
			if(CURRENT_LOCK.equals(firstNode)){//当前节点和最小锁比较，如果相同，则获取锁成功
				return true;
			}
			if(!lessthanMe.isEmpty()){
				WAIT_LOCK = lessthanMe.last();//获取比当前节点更小的最后一个节点，设置给WAIT_LOCK
			}
		} catch (Exception e) {
			// TODO: handle exception
			e.printStackTrace();
		}
		return false;
	}

	@Override
	public boolean tryLock(long arg0, TimeUnit arg1) throws InterruptedException {
		// TODO Auto-generated method stub
		return false;
	}

	@Override
	public void unlock() {
		System.out.println(Thread.currentThread().getName()+"->"+CURRENT_LOCK+"释放锁");
		try {
			zk.delete(CURRENT_LOCK, -1);
			CURRENT_LOCK = null;
			zk.close();
		} catch (Exception e) {
			// TODO: handle exception
		}
	}
	
	
}

 

测试类

package com.lf.zookeeper.lock;

import java.io.IOException;
import java.util.concurrent.CountDownLatch;

public class LockDemo {
	
	public static void main(String[] args) throws IOException {
		CountDownLatch countDownLatch = new CountDownLatch(10);
		for (int i = 1; i <= 10; i++) {
			new Thread(()->{
				try {
					countDownLatch.await();
					DestributeLock destributeLock = new DestributeLock();
					destributeLock.lock();
				} catch (Exception e) {
					e.printStackTrace();
				}
			},"Thread-"+i).start();
			countDownLatch.countDown();
		}
		System.in.read();
	}
}

 运行结果

 　　

Thread-4->/locks/0000000072,尝试竞争锁！
Thread-5->/locks/0000000073,尝试竞争锁！
Thread-9->/locks/0000000074,尝试竞争锁！
Thread-8->/locks/0000000075,尝试竞争锁！
Thread-10->/locks/0000000076,尝试竞争锁！
Thread-2->/locks/0000000077,尝试竞争锁！
Thread-6->/locks/0000000078,尝试竞争锁！
Thread-3->/locks/0000000079,尝试竞争锁！
Thread-7->/locks/0000000080,尝试竞争锁！
Thread-1->/locks/0000000071,获取锁成功！
Thread-4->等待/locks/locks/0000000071释放锁
Thread-5->等待/locks/locks/0000000072释放锁
Thread-9->等待/locks/locks/0000000073释放锁
Thread-8->等待/locks/locks/0000000074释放锁
Thread-10->等待/locks/locks/0000000075释放锁
Thread-2->等待/locks/locks/0000000076释放锁
Thread-6->等待/locks/locks/0000000077释放锁
Thread-3->等待/locks/locks/0000000078释放锁
Thread-7->等待/locks/locks/0000000079释放锁

 手动触发watcher事件，释放锁，delete /locks/0000000071

出现     Thread-4->/locks/0000000072,获取锁成功！

三、基于curator的实现分布式锁

　　代码

　　

package com.lf.zookeeper.lock;

import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.locks.InterProcessMutex;

public class CuratorLockDemo {
	
	public static void main(String[] args) {
		CuratorFramework curatorFramework = CuratorFrameworkFactory.builder().build();
		InterProcessMutex interProcessMutex = new InterProcessMutex(curatorFramework, "/locks");//关注节点
		try {
			interProcessMutex.acquire();//获取锁
		} catch (Exception e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}
}