集群环境中使用Redis实现分布式锁两种方式

左手的ㄟ右手 2022-01-15 17:35 253阅读 0赞

# 一、介绍 #

互联网的应用场景中，为了支持高并发的请求，服务都是执行的分布式部署，相同的任务可以在集群中不同的服务器上执行，并且现在的服务容器都是支持多线程，相同的任务也可能会被同一个容器多次执行，都要求执行结果都满足**幂等性**的设计原则。

分布式锁，就是为了确保在分布式的环境下，相同任务只会执行成功的执行一次，后续的执行不会对这些已经产生了变化的业务再次产生影响。

分布式锁的实现有不少的方式，如：

1.  使用RDBMS数据库本身的表锁或行锁特性；
2.  使用Redis做为分布式锁；
3.  使用Zookeeper做为分布式锁；

使用RDBMS数据库做为锁不是笔者要讨论的范畴，因为其本身的特性，不太符合在高并发下锁的应用场景，这里以Redis作为分布式锁做为介绍。

# 二、Redis #

Redis本身有一些命令组支持原子性的操作，如getset、setns，这些命令可以用于分布式锁的场景中。

## **1、使用getset作为分布式锁控制实现** ##

getSet本身是支持原子性的，在写入新值的同时会返回旧的值，用这个写入新值并获取旧值做为分布式锁的控制实现，如果返回的值不为空，那就说明前面已经有其它线程（这里的其它线程可以指当前容器中当前服务的其它线程，也指由部署在其它服务器上的应用中的线程）修改了该值，则可以认为已经有线程在对该请求正在处理，因而可以放弃后面的处理逻辑。

可以将当前系统的时间作为分布式锁key的值，后续其它线程的请求时，将其请求的时间与获取到的锁对应的key的旧值进行比较，比较是否已经超过了一定的时间控制阀值，如果超过了**则可以认为**（这里存在误判的可能性，因为后续逻辑的数据处理，恰好超过了这个时间比较阀值，就会导致重复执行，因而这里时间控制阀值要设置的比较合理，另外也需要合适的熔断机制用于保证）前面的交易处理失败（如服务恰好在设置了用于分布式锁的key后，立即就挂了，没有执行到后面的删除操作）导致用于分布式锁的key没有被删除掉，可以继续处理该请求的后续交易逻辑。

这个是非常轻量级的事务控制，不会对Redis产生外部事务（应用与Redis之间的交互事务），只是需要对Redis多进一次getset操作，流程图如下：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZlbmdsaWJpbmc_size_16_color_FFFFFF_t_70][]

其中蓝色部分表示获取锁的逻辑。

Java代码实现如下：

@Resource
    private RedisTemplate<String, String> redisTemplate;
    // 超时时间，以毫秒为单位
    private final long timeout = 2000;
    
    @Test
    public void testLock() {
    	// 用于判断交易唯一性和合法性的Token，在交易执行之前先保存在服务端，
    	// 并且下发给客户端，客户端会在执行交易之前把Token带上，没带Token的
    	// 请求、Token不存在的服务端的请求、Token不正确的请求都视为非法请求
    	String token = "...";
    	String key = MD5Util.md5Of32(token);
    	String lockKey = new StringBuilder(key).append("_lock").toString();
    	boolean isGetKey = getLock(lockKey);
    	if (!isGetKey) {
    		log.warn("当前交易正在被处理中");
    	}
    	boolean handleSuccess = false;
    	try {
    		log.info("处理交易开始");
    		String storedToken = redisTemplate.opsForValue().get(key);
    		// 判断Token是否存在且合法
    		if (!token.equals(storedToken)) {
    			log.warn("指定的Token不存在.");
    			return;
    		}
    		handleSuccess = true;
    		log.info("处理交易结束");
    	} catch (Exception e) {
    		log.info("处理发生异常", e);
    	} finally {
    		if (handleSuccess) {
    			// 限制了单个Token只能够执行一笔记交易，因而执行成功后将其删除
    			List<String> keys = new ArrayList<String>();
    			keys.add(key);// 限制了单个Token只能够执行一笔记交易，因而执行成功后将其删除
    			keys.add(lockKey);// 用于表示锁的key删除，表示释放掉锁
    			redisTemplate.delete(keys);
    		} else {
    			// 删除用于锁定的key
    			redisTemplate.delete(lockKey);
    		}
    	}
    }
    
    /**
     * 原理是从redis中获取到的lockKey的值是不是存在，如果不存在表示写入的是当前值，表示锁获取成功；
     * 如果获取到的值存在，再判断是否已经超过了指定的期限，如果超过了指定的期限，则认为锁获取成功，否则认为锁获取失败；
     * 
     * @param lockKey 用于获取锁定的key
     * @return true表示获取到锁，false表示未获取到锁
     */
    public boolean getLock(String lockKey) {
    	long now = System.currentTimeMillis();
    	// Redis的GetSet返回的值必须是字符串，否则会抛异常，因而将其转换为字符串
    	String nowTime = String.valueOf(now);
    	String oldTime = null;
    	// 判断用于锁定的key是否已经被设置了值，如果被设置了值，则用于控制后续的处理逻辑不再进行
    	if ((oldTime = redisTemplate.opsForValue().getAndSet(lockKey, nowTime)) != null) {
    		// 检查锁lockKey的值是不是超过了设定的时间，如2秒钟，没有超过则返回，不继续处理后续的任务；
    		// 注：这个逻辑有个问题，就是客户端在2秒钟之内不停的重试，就永远不会进入到后面的处理环节。
    		// 不过针对正常的业务请求这个是可以约定的，针对非正常的请求，被拦截也很正常，所以这个问题不是问题。
    		if (now - Long.parseLong(oldTime) < timeout) {
    			return false;
    		}
    		return true;
    	}
    	return true;
    }

## **2、使用setnx作为分布式锁控制实现** ##

setnx和getset的执行逻辑不同，getset是设置新值并返回旧值，setnx如果存在旧值时可以通过参数控制不设置值并返回0,不存在旧值时才设值并返回1。

二者处理流程上都是相同的，不同之处在于获取锁的实现，setnx的实现逻辑如下：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZlbmdsaWJpbmc_size_16_color_FFFFFF_t_70 1][]

其中绿色部分为setnx获取锁的逻辑，这个和getset是不同的实现逻辑。

setnx获取锁的Java代码实现如下：

public boolean getLock(String lockKey) {
    	RedisConnection connection = redisTemplate.getConnectionFactory().getConnection();
    	JedisCommands commands = (JedisCommands) connection.getNativeConnection();
    	boolean con = false;
    	do {
    		long now = System.currentTimeMillis();
    		// Redis的GetSet返回的值必须是字符串，否则会抛异常，因而将其转换为字符串
    		String nowTime = String.valueOf(now);
    		con = false;
    		// 返回1表示锁获取成功，返回0表示锁取失败
    		String result = commands.set(lockKey, nowTime, "NX", "PX", expire);
    		if ("1".equals(result)) {
    			return true;
    		} else {
    			String oldTime = redisTemplate.opsForValue().get(lockKey);
    			if (null != oldTime) {
    				// 检查锁lockKey的值是不是超过了设定的时间，如2秒钟，如果超过了则继续尝试获取锁，
    				// 直到获取到锁，或者数据未超期时退出，循环判断可以解决死锁的问题
    				if (now - Long.parseLong(oldTime) >= expire) {// 数据已经过期了
    					con = true;
    				}
    			}
    		}
    	} while (con);
    	return false;
    }

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZlbmdsaWJpbmc_size_16_color_FFFFFF_t_70]: /images/20220115/2b1473b0e2514719909abe5d6cb8b209.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZlbmdsaWJpbmc_size_16_color_FFFFFF_t_70 1]: /images/20220115/92910901a1934c31a89456acf4273a54.png