基于Zookeeper和Redis实现分布式锁

一、前言

思考：怎么样设计一个分布式锁？它有哪些核心的关键点呢？

• ① 抢锁的设计，实现跨 JVM 的互斥
• 要求：不同的节点（应用）去抢锁对象的时候，同一时刻有且只有一个人能抢成功
• 锁对象（名称）的选择，一般都是根据具体业务来选择，比如说：文件上传如果使用分布式锁，那么文件 MD5 做锁对象。
• 支持互斥功能的中间件，比如：数据库的悲观锁、Redis 的 setnx(key, value)、Zookeeper 的目录唯一性等，这些特性都能实现互斥效果。

• ②堵塞的设计，抢不到锁时应该怎么办呢？
• 方案一：循环，抢到不到锁的线程，不断的轮询，直到锁被释放并且自己抢成功为止
• 方案二：堵塞 + 通知，如果抢到不到锁的线程，则该线程处于堵塞状态（比如：J.U.C 下面的几个常见工具类，CountDownLatch、Semaphore、CyclicBarrier 有可以实现堵塞和唤醒的功能），锁被释放时，通知客户端，唤醒等待的线程去抢锁。

• ③释放锁的设计，执行完成之后进行锁的释放
• 正常情况下，执行完成业务之后，释放锁，并且通知其他线程去抢锁
• 要求 1：不能释放别人的锁，如果释放别人的锁，则会出现干扰的情况
• 要求 2：不能出现死锁，如果抢到锁的进程挂掉了，无法释放锁了，那么其他线程只能一直处于堵塞状态。

二、Zookeeper 分布锁

2.1、原理分析

思考：为什么 Zookeeper 适合做分布式锁呢？

• 特性一：临时节点（推荐）
• 某个目录下，不能出现名称相同的节点，因此多个线程同时创建某个目录（锁对象）时，有且只有一个创建成功
• 连接断开时，临时节点自动删除，可以完美解决死锁的问题
• Zookeeper 可以 Watcher（订阅）某个节点，只要该节点发生变化（比如：删除），则其他客户端会收到通知

• 特性二：有序临时节点
• 如果多个线程抢锁时，那么按照抢锁的顺序创建有序临时节点（一个线程对应一个节点，节点名称是有序递增，如：xxxxx-0001，xxxxx-0002）
• 每个线程判断当前节点（n）是否是最小值，如果是最小值则表示抢到所有执行权限；否则监听自己的上一个节点（n-1），上一个节点删除之后，则监听该节点的客户端收到通知
• 连接断开时，临时节点自动删除，可以完美解决死锁的问题
• Zookeeper 可以 Watcher（订阅）某个节点，只要该节点发生变化（比如：删除），则其他客户端会收到通知

思考：Zookeeper 是否会释放别人的锁呢？

答案：不会

• 比如：网络抖动，让抢到锁的线程断开了一下，其他线程抢锁，等原来线程恢复之后是否会释放别人的锁呢？看似会，其实不会，因为 Zookeeper 的 sessionTimeout 指的是会话过期时间，一旦连接断开，临时节点不会立马被删除，判断 sessionTimeout 期间之内连接是否恢复，如果不恢复才删除临时节点，其他线程才有机会抢锁，如果超过 sessionTimeout 没有恢复，那么它想恢复，它必须重新抢锁了，因此不存在释放别人的锁的情况。

思考：Zookeeper 会不会出现死锁呢？

答案：不会

• 因为一旦抢到锁的线程宕机了，那么连接自动断开，自动释放锁

• 特性一实现分布式锁相对简单，Zookeeper 服务端不必要创建大量的临时节点，但是如果释放锁的时候，那么会存在大量的线程再次抢锁的情况。

• 特性二虽然不会存在释放锁的时候大量线程抢锁，但是需要创建大量的临时节点，并且对临时节点进行排序计算

提示：curator-recipes 框架，已经帮我们封装好了 Zookeeper 的分布式锁的实现，但是更加推荐自己手工实现一遍，才能真正的掌握其原理。

2.2、创建相关文件

netdisk-service-provider
	|-- com.micro.lock
	|  |-- Lock.java				（定义接口类）
	|  |-- LockZookeeper.java		（Zookeeper锁实现类）
	|  |-- LockRedis.java			（Redis锁实现类）
	|  |-- LockContext.java			（策略类）

2.3、定义接口类

public interface Lock {
	//获取锁
	public void getLock(String lockname);
	//释放锁
	public void unLock(String lockname);
}

2.4、接口实现类

//@Component，注意这里不能加该注解
public class LockZookeeper implements Lock {
	private CountDownLatch latch = new CountDownLatch(1);
	private static volatile ZkClient zkClient =null;
	private static String ROOT="/locks";

    //单例模式，保证zkClient是单例
	private LockZookeeper(){}
	public static LockZookeeper getInstance(String url){
        if(zkClient==null){
            synchronized (LockZookeeper.class) {
                if(zkClient==null){
                    zkClient = new ZkClient(url, 30000);
                    boolean b=zkClient.exists(ROOT);
                    if(b==false){
                        zkClient.create(ROOT, "", CreateMode.PERSISTENT);
                    }
                }
            }
        }
		return new LockZookeeper();
	}

	//加锁
	@Override
	public void getLock(String lockname) {
        try{
            //抢锁：创建节点
            zkClient.createEphemeral(ROOT+"/"+lockname);
        }catch(Exception e){
             //抢锁失败：堵塞状态
        	waitLock(lockname);
        	// 重写获取锁的资源
        	getLock(lockname);
        }
	}

	// 释放锁
	@Override
	public void unLock(String lockname) {
		zkClient.delete(ROOT+"/"+lockname);
	}
    //堵塞
	private void waitLock(String lockname) {
        //创建监听事件
		IZkDataListener zkDataListener = new IZkDataListener() {
            //节点删除事件
			public void handleDataDeleted(String path) throws Exception {
				countDownLatch.countDown();//唤醒
			}
		};
		//监听事件
		zkClient.subscribeDataChanges(ROOT+"/"+lockname, zkDataListener);

        //如果节点存在，则创建信号量，并且堵塞
		if (zkClient.exists(ROOT+"/"+lockname)) {
			countDownLatch = new CountDownLatch(1);//重新创建
			countDownLatch.await(); //堵塞
		}

		//唤醒之后，会执行这里，取消订阅监听事件
		zkClient.unsubscribeDataChanges(ROOT+"/"+lockname, zkDataListener);
	}
}

2.5、策略类

public class LockContext {
	private String locktype;
	private String host;
	private Lock lock;

	public LockContext(String locktype,String host){
		this.locktype=locktype;
		this.host=host;
	}

	//获取锁
	public void getLock(String lockname){
		if("Zookeeper".equals(locktype)){
            //注意：这里没有使用Spring容器管理!!!!
			lock=LockZookeeper.getInstance(host);

		}else if("Redis".equals(locktype)){
			lock=LockRedis.getInstance(host);

	    }else{
	    	throw new RuntimeException("找不到标识locktype=="+locktype);
	    }

		lock.getLock(lockname);
	}

	//释放锁
	public void unLock(String lockname){
		lock.unLock(lockname);
	}
}

• 通过策略模式来调用真正的实现类，跟 FastDFS 章节一样

2.6、策略类调用

LockContext lc=new LockContext("zookeeper","192.168.1.8:2181");
String lockname="xxxx";
lc.getLock(lockname);
lc.unLock(lockname);

思考：策略类为什么不能交给 Spring 容器进行管理呢？

2.7、容易犯错的细节解析

public class LockZookeeper implements Lock {
    private static String ROOT="/locks";
	private CountDownLatch latch = new CountDownLatch(1);

    //这里不用static修饰，每个实例单独拥有
	private ZkClient zkClient =null;

    //构造函数
	public LockZookeeper(String url){
        //每次创建实例，都创建一个ZkClient，最后导致连接数量过多
		zkClient = new ZkClient(url, 30000);
        boolean b=zkClient.exists(ROOT);
        if(b==false){
            zkClient.create(ROOT, "", CreateMode.PERSISTENT);
        }
	}
}

• 缺点：高并发情况下，客户端对象过多导致连接 Zookeeper 的长连接过多，直接卡死

//分布式锁实现类
@Component
public class LockZookeeper implements Lock {
    //信号量（J.U.C并发编程里面的工具类）
	private CountDownLatch latch = new CountDownLatch(1);

    //......省略
}

//策略类
@Component
public class LockContext{
    @Autowired
    private Lock lockZookeeper;

    public void getLock(String lockname){
       lockZookeeper.getLock(lockname);
    }
}

//业务调用
@Autowired
private LockContext lockContext;
lockContext.getLock("xxxx");

思考：为什么不能通过 @Component 注解把策略类和实现类交给 Spring 容器管理呢？

• Spring 默认是创建单例的，如果该实例里面的属性变量只读则线程安全；但是如果是涉及修改等特殊处理，将会出现线程不安全的问题。

• CountDownLatch 是 LockZookeeper 类的属性（用来做信号量），它是用来控制线程抢锁失败时，线程堵塞的。如果是单例情况下，所有的线程都共用一个信号量那么将会出现不安全情况，必须一个实例对应一个信号量。

• 优化：除了使用使用多实例之外，其实可以借助单例 +ThreadLocal 来解决也是可以的

三、Redis 分布锁

3.1、原理分析

思考：为什么 Redis 适合做分布式锁呢？

• setnx (key,value) 的特性是，如果 key 存在则无法保存，并且 Redis 是单线程的，因此能保证互斥性

• 可以给 key 设置过期时间，避免死锁的发生

• 问题 1：释放别人的锁，举例：线程 A 卡死了，一段时间之后 Redis 自动释放锁被线程 B 抢到，等线程 A 恢复过来继续释放锁的时候，把线程 B 的锁给释放了。
• 解决：将 value 赋值为 UUID，代表加锁的客户端请求标识，那么在客户端在释放锁的时候就可以进行校验是否是同一个客户端（注意：如果 value 是固定值，则是有问题的！！）

• 问题 2：如果释放锁时，每次都判断 value 是否一致，如果一致则删除 key，这个是两个步骤去执行，很容易出现原子性问题
• 举例：客户端 A 加锁，一段时间之后客户端 A 释放锁，判断 value 一致，准备执行 jedis.del()，锁突然过期了，此时客户端 B 尝试加锁成功，然后客户端 A 再执行 del () 方法，则将客户端 B 的锁给解除了
• 解决：Redis 执行 Lua 脚本，由于 Redis 是单线程的，Lua 脚本可以保证原子性

• 为了保证原子性，需要做以下操作

1. 根据 key 去 redis 获取对应的 value

2. 获取到的 value 和本地的 value 做比较，目的是防止释放别人的锁

3. 为了防止高并发下，数据安全问题，以上两步必须在一次 Redis 请求之内完成

3.2、代码实现

public class LockRedis{
    public void getLock(Jedis jedis,String key,String value,int expireTime) {
        Long result = jedis.setnx(key, value);//设置值
        if (result == 1) {
            //bug1：如果这里宕机了，则无法设置过期时间，那么就可能出现死锁
            jedis.expire(lockKey, expireTime);
        }
    }
    public static void unLock(Jedis jedis, String lockKey, String value) {
        if (value.equals(jedis.get(lockKey))) {
            //bug2：比如客户端A加锁，一段时间之后客户端A解锁，在执行`jedis.del()`之前，锁突然过期了，此时客户端B尝试加锁成功，然后客户端A再执行del()方法，则将客户端B的锁给解除了
            jedis.del(lockKey);
        }
    }
}

• Redis 分布执行则容易出现原子性问题，参考上面的问题1、问题2

public class LockRedis{
    //加锁
    public boolean getLock(Jedis jedis,String key,String value,int expireTime){
        //保存的时候，同时设置过期时间
        String result = jedis.set(key, value, "NX", "PX", expireTime);
        if ("OK".equals(result)) {
            return true;
        }
        return false;

    }

    //释放锁
    private static final Long RELEASE_SUCCESS = 1L;
    public static boolean unLock(Jedis jedis, String key, String value) {

        //Lua脚本
        String lua = "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end";
        Object result = jedis.eval(lua, Collections.singletonList(key), Collections.singletonList(value));

        if (RELEASE_SUCCESS.equals(result)) {
            return true;
        }
        return false;

    }
}

• eval () 方法是将 Lua 代码交给 Redis 服务端执行
• eval 命令执行 Lua 代码的时候，Lua 代码将被当成一个命令去执行，并且直到 eva l 命令执行完成，Redis 才会执行其他命令

• 脚本的含义：首先获取锁对应的 value 值，检查是否与 value 相等，如果相等则删除锁

3.3、Redis 分布式锁的弊端

1. Redis 分布式锁一般是实现方案都是基于 Redis 单节点的，如果是 Redis Cluster 模式下那么分布式锁实现就会比较复杂

2. 抢不到锁的时候，需要不断的轮询，无法像 Zookeeper 那样订阅节点的名称，比较浪费性能

3. 如果抢到锁的客户端宕机了，不能立马释放锁，无法像 Zookeeper 那样立马释放锁

四、应用场景

• 案例一：解决事务并发问题
• 场景：秒杀商品扣减库存，很多人同时购买同一个商品，MySQL 就会出现事务并发问题
• 危害：由于判断库存是否小于0 和扣减库存两个步骤是分开的，如果高并发情况下很容易超卖等问题。
• 解决：可以使用分布式锁来解决，使用商品 ID 作为锁的名称即可。

• 案例二：特殊业务解决方案
• 场景：网盘系统的上传功能，如果多个人同时上传同一份文件，网盘系统只保留一份，避免重复存储
• 好处：使用文件MD5 作为锁名称，分布式锁可以很好的解决文件重复上传的问题

• 案例三：定时器集群部署
• 场景：项目 A 实现了定时器功能，此时项目 A 需要做集群部署
• 危害：每次到点的时候，不同服务器上面的项目 A 的定时器同时执行，出现跑重的情况。
• 解决：使用分布式锁去解决，谁抢到锁，谁有权限执行。

• 案例四：过期监听（后面章节讲到）
• 场景：集群模型下，监听 Redis 的 key 过期，并且做业务处理
• 危害：集群的每个节点都会收到事件，都会去做处理，导致重复处理
• 解决：可以使用分布式锁来解决，使用 key 作为锁的名称即可。

五、小结