Redis分布式锁的正确使用姿势

前言

分布式锁在日常开发中，用处非常的多。包括但不限于抢红包，秒杀，支付下单，幂等，等等场景。
分布式锁的实现方式有多种，包括redis实现，mysql实现，zookeeper实现等等。而其中redis非常适合作为分布式锁使用，并且在各个公司都大规模的使用。

本文将
由浅入深的
探究Redis分布式锁的实现，最终实现一个可工业使用的Redis分布式锁。欢迎大家一步一步跟读，一起学习一起进步。

什么是分布式锁

分布式锁其实就是，控制分布式系统不同进程共同访问共享资源的一种锁的实现。如果不同的系统或同一个系统的不同主机之间共享了某个临界资源，往往需要互斥来防止彼此干扰，以保证一致性。

举一个最简单的例子。有一个数据库字段status=0，表示初始状态。
只有在status=0初始状态
。才能修改这个值。现在有两个人，张三和李四。

• 张三，发起请求将status=0 修改为 status=1
• 李四，发起请求将status=0 修改为 status=2

因为只有status=0才会修改，代码在修改之前都会去查询status的值，并且判断是否为0。如果为0才会去更新，不为0，则拒绝更新。这其实就是一个幂等的实现。

• 假如没有分布式锁。短时间内请求两次，
  此时两次都获取status=0
  ，一个修改成了1，一个修改成了2。破坏代码逻辑，有问题
• 假如加上分布式锁。短时间内请求两次，
  只有第一笔请求结束之后，第二笔才会执行
  。也就是第二笔获取status，只能获取到最新的值，比如status=1，则不修改。

Redis分布式锁方案一：SETNX （不推荐）

public String lockA(String key) {
	String val = UUID.randomUUID().toString();
	// set k  v  nx  如果不存在则设置成功，如果存在则设置失败
	boolean success = stringRedisTemplate.opsForValue().setIfAbsent(key, val);
	if (success) {
		log.info("lock success");
		try {
			// do something
		} finally {
			stringRedisTemplate.delete(key);
		}

	} else {
		log.info("lock fail");
	}

	return "lockA";
}

这个方案有一个最大的问题就是，如果线程A获取锁成功，并没有设置过期时间。那么如果此时doSomething里面是一个死循环或者程序在期间重启了，就会导致这个锁就不会被释放，那么
别的线程永远获取不到锁啦
。这个问题非常严重，对业务影响极大。不推荐使用。

Redis分布式锁方案二：SETNX + expire （不推荐）

那既然没有过去时间，我就设置一个过期时间不就行了，代码如下。

public String lockB(String key) {
	String val = UUID.randomUUID().toString();

	// set k  v  nx  如果不存在则设置成功，如果存在则设置失败
	boolean success = stringRedisTemplate.opsForValue().setIfAbsent(key, val);
	stringRedisTemplate.expire(key, 60, TimeUnit.SECONDS);
	if (success) {
		log.info("lock success");
		try {
			// do something
		} finally {
			stringRedisTemplate.delete(key);
		}

	} else {
		log.info("lock fail");
	}

	return "lockB";
}

这个方案2和方案1有同样的问题。setnx 和 expire不是一个原子执行。在获取锁成功之后，准备执行expire的时候，程序重启，也会导致同样方案1的问题，此处不再赘述。不推荐使用。

Redis分布式锁方案三：SET EX NX （不推荐）

那既然不是原子性，我们就用原子性就好了。从
redis 2.6.12
开始，set方法支持 set ex nx

public String lockB(String key) {
	String val = UUID.randomUUID().toString();

	// set k  v ex  nx  如果不存在则设置成功，如果存在则设置失败
	boolean success = stringRedisTemplate.opsForValue().setIfAbsent(key, val, 60, TimeUnit.SECONDS);
	if (success) {
		log.info("lock success");
		try {
			// do something
		} finally {
			stringRedisTemplate.delete(key);
		}

	} else {
		log.info("lock fail");
	}

	return "lockB";
}

从方案三开始，此代码就比较有健壮性了。有部分公司使用的就是方案三，但仍然存在两个问题

• doSomething还没执行完，锁过期就被自动释放了。那么其他线程就可以获取此锁了。就会导致此代码块可能被多个线程执行。当然使用的时候可以把过期时间设置大一点，比如60分钟，3个小时等等，但总归不太好。
• 线程A获取锁，没执行完成，锁过期了。此时线程B获取锁执行了。然后A执行完成去释放锁的时候，但他释放的是线程B获取的锁，此时是有问题的，并且问题还不小。同样不推荐使用。

Redis分布式锁方案四： （推荐）

既然时间太短，我就设置过期时间长一点。既然会被误删，我们就判断一下。代码如下

public String lockD(String key) {
	String val = UUID.randomUUID().toString();

	// set k  v  nx  如果不存在则设置成功，如果存在则设置失败
	boolean success = stringRedisTemplate.opsForValue().setIfAbsent(key, val, 60, TimeUnit.MINUTES);
	if (success) {
		log.info("lock success");
		try {
			// do something
		} finally {
			if (val.equals(stringRedisTemplate.opsForValue().get(key))) {
				stringRedisTemplate.delete(key);
			}

			// String script = "if redis.call('get',KEYS[1]) == ARGV[1] then return redis.call('del',KEYS[1]) else return 0 end";
			// DefaultRedisScript<Long> redisScript = new DefaultRedisScript<>();
			// redisScript.setScriptText(script);
			// redisScript.setResultType(Long.class);
			// return stringRedisTemplate.execute(redisScript, Collections.singletonList(key));

		}

	} else {
		log.info("lock fail");
	}

	return "lockD";
}

大部分公司，我相信使用的都是方案四。方案四正常来说，在使用过程中极大概率不会出现任何问题，除非你们的量非常的大。但其仍有问题，finally删除锁的那块不是原子性。

比如线程A获取锁成功uuid=123, 释放成功。线程B获取锁，uuid=456,锁过期，自动释放。
此时A再次获取锁，uuid=456（恰巧是456,概率非常低）。那么A就会释放B的锁。因此为了更加严谨一点，我们使用lua脚本来保证，判断+删除的原子性。

方案四已经符合绝大多数公司的使用了，但其不好估计的过期时间，以及释放的原子性，仍 概率性的存在问题。所以社区为了解决此问题，有了以下方案。

Redis分布式锁方案五： Redission方案 （推荐）

Redisson官网介绍： Easy Redis Java client with features of an in-memory data grid（易于使用的 Redis Java 客户端，具备内存数据网格的特性）

Redisson 是一个基于 Java 的 Redis 客户端库，它提供了一系列的高级功能，使得在 Java 应用程序中使用 Redis 变得更加方便和强大。Redisson 的目标是充分利用 Redis 的各种特性，同时提供易于使用的 Java 接口。

RedissonClient
是 Java 中 Redisson 库提供的一个接口，它封装了对 Redis 数据库的各种操作，提供了丰富的方法来与 Redis 进行交互。Redisson 是一个在 Redis 的基础上实现的 Java 内存数据网格（In-Memory Data Grid）。它不仅提供了对基本数据结构的操作，还提供了分布式的 Java 对象和服务，例如
分布式锁
、集合、映射、发布/订阅、计数器等。

我们这次使用到的是redission的分布式锁。

// 获取锁  
public String lockE(String key) {

	// 获取锁
	RLock lock = redissonClient.getLock(key);
	try {
		// 获取锁。此处30s不是指执行30s，而是获取锁的超时时间
		if (lock.tryLock(30, TimeUnit.SECONDS)) {
			log.info("lock success");
		}
	} catch (Exception e) {
		
	} finally {
		if (lock != null && lock.isHeldByCurrentThread()) {
			lock.unlock();
		}
	}

	return "lockE";
}

此方案基本适用于99.99%的公司，当然可能会出现Redlock的问题，此处不过多讨论，感兴趣的同学可以网上自行搜索。

只要线程加锁成功，默认过期时间是30s。后台会自动启动一个
watch dog
看门狗，它是一个后台线程，会每隔10秒检查一下，如果线程还持有锁，那么就会不断的延长锁key的生存时间。因此，Redisson就是使用Redisson解决了
锁过期释放，业务没执行完
问题。

具体Redission常见问题，以及源码分析，可以详见：
Redis分布式锁实现Redisson 15问（面试常问）

最后

本文由浅入深的介绍了分布式锁。解释了为什么大部分公司用的都是方案四以及方案五的实现，而不是方案1,2,3。我们需要知道每个方案的优劣势，从而选出最适合我们业务的一种技术方案，这是每个架构师都应该具备的一种能力。