Redis分布式锁

1.分布式锁的方案

2.靠谱的分布式锁有的特征

• 互斥性：任意时刻，只有一个客户端能持有锁
• 锁超时释放：持有锁超时，可以释放，防止不必要的资源浪费，也可以防止死锁
• 可重入性：一个线程如果获取了锁之后，可以再次对其请求加锁
• 高性能和高可用：加锁和解锁需要的开销尽可能低，同时也要保证高可用，避免分布式锁失效。
• 安全性：锁只能被持有的客户端删除，不能被其它客户端删除

3.redis分布式锁的方案

3.1 setnx+expire

这种方式是先通过setnx操作进行加锁，加锁成功后通过expire操作进行设置过期时间。但是由于两个命令是分开的，不是原子操作。因此可能会出现在加锁成功后，还没设置过期时间的时候进程重启了，那么这个锁就会一直存在了。具体的伪代码如下：

if (jedis.setnx(id, value) == 1) { //加锁成功
	expire(id, 100); //设置锁过期时间
	try {
		//业务操作
	} catch() {
		//业务处理异常
	} finally {
		jedis.del(id); //释放锁
	}
}

3.2 setnx+value值是(系统时间+过期时间)

这种就是一个setnx命令，在设值的时候把value值设为系统时间加过期时间，如果加锁成功，则过期时间就为value的值，加锁失败，则需要校验一下value的值是否已过期，重新设置过期时间，返回加锁成功。伪代码如下：

{
	String expiresStr = String.valueOf(System.currentTimeMillis() + 60000); //当前时间+过期时间
	//当前锁不存在，则返回加锁成功
	if (jedis.setnx(id, expiresStr) == 1) {
		return true;
	}
	//如果锁已存在，获取锁的过期时间
	String expireTime = jedis.get(id);
	//判断锁是否已过期
	if (StringUtils.isNotBlank(expireTime) && Long.parseLong(expireTime) < System.currentTimeMillis()) {
		//锁已过期，获取旧的过期时间设置新的过期时间
		String oldExpireTime = jedis.getSet(id, expiresStr);
		//防止多线程并发，只有一个线程的设置值与当前值相同，才可以加锁
		if (StringUtils.isNotBlank(oldExpireTime) && oldExpireTime.equals(expireTime)) {
			return true;
		}
	}
	return false;
}

这种方式可以解决setnx与expire操作不是原子操作的问题。但该方案同样有缺点：

• 过期时间是客户端生成的，因此每个客户端的时间必须同步
• 如果有多个线索执行jedis.getSet()，最终只有一个加锁成功，但是可能会导致其它线程覆盖加锁线程的锁过期时间。
• 锁可能会被别的线程解锁。

3.3 使用Lua脚本

使用Lua脚本能够保证setnx+expire原子性

String luaLock = "if redis.call('setnx', KEYS[1],ARGV[1]) == 1 then redis.call('expire', KEYS[1], ARGV[2]) return 1 else return 0 end";
Object result = jedis.eval(luaLock, Collections.singletonList(id), Colections.singletonList(valuse));
return result.equals(1L);

这种方式也是有缺点的，学习成本比较高，使用lua脚本应该要掌握一定的lua语法。无法进行代码复用。

3.4 set的扩展命令

另外还可以通过set的扩展参数*(SET key value [EX seconds] [PX milliseconds] [NX|XX])来实现，其也是原子性的。

• NX：表示key不存在的时候，才能set成功。
• EX seconds：设定key的过期时间，时间单位是秒
• PX milliseconds：设定key的过期时间，单位为毫秒
• XX：仅当key存在时设置值
  伪代码如下：

if (jedis.set(id, value, "NX", "EX", 100s) == 1) {
	try {
		//业务逻辑
	} catch() {
		//异常时的操作
	} finally {
		jedis.del(id);
	}
}

这种方式也会存在一些问题：
线程A持有锁过期释放掉了，但是业务还没有处理完成，此时有另一个线程B获取到锁，执行业务逻辑，但是此时线索A执行完业务逻辑之后会释放掉线程B持有的锁。

3.5 set的扩展命令+校验唯一的随机值，再删除

为了保证只会删除自己的锁而不会误删其它线程的锁，我们可以为value设置一个唯一的值，在删除之前进行校验一下是不是当前线程的锁就行了。这里为了保证校验是原子性的，可以使用lua脚本的方式。伪代码如下：

if (jedis.set(id, value, "NX", "EX", 100s) == 1) {
	try {
		//业务逻辑
	} catch() {
		//异常时的操作
	} finally {
		deleteLock(id, value);
	}
}

public boolean deleteLock(String id, String value) {
	String lua = "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end";
	return jedis.eval(lua, Collections.singletonList(id), Collections.singletonList(value)).equals(1L);
}

3.6 Redisson框架

上面的方式也会存在当设置的锁的已经过期释放了，但是业务代码还没执行完的问题。那么此时就可以通过Redisson框架实现，其会启动一个watchDog用来给锁进行延时。需要注意的是使用此方法通过tryLock方法进行加锁时不能设置过期时间，设置的话就不会启动watchDog。伪代码如下：

public void testLock(String key) {
    RLock lock = null;
    try {
        // 读取配置文件
        Config config = Config.fromYAML(new File("redis-file.yaml"));
        RedissonClient redissonClient = Redisson.create(config);
        lock = redissonClient.getLock(key);
        // 不设置锁的过期时间，启动watchDog机制
        boolean b = lock.tryLock(10, TimeUnit.SECONDS);
        if (b) {
            // 加锁成功后的业务逻辑
        }
    } catch (Exception e) {
        log.error("加锁失败");
    } finally {
        if (Objects.nonNull(lock)) {
            lock.unlock();
        }
    }
}

redisson框架的底层是通过lua脚本进行加锁与解锁的，保证了其加锁时的原子性。

3.7 RedLock

对于集群的情况，如果线索成A在主节点拿到锁，但是加锁的key还没有同步到从节点，此时主节点发生了宕机，一个从节点升为主节点。线程B可以再次获取同个key的锁。这样就会导致锁不安全了。
而为了解决这个问题则可以使用redisson的redlock算法。其核心思想是部署多个Redis主节点，这些节点完全互相独立，不存在主从复制或者其它集群协调机制；依次尝试从N个Master实例使用相同的key和随机值获取锁，当至少有N/2 + 1个redis实例加锁成功后，才是加锁成功。
其实现步骤如下：

• 获取当前时间，以毫秒为单位
• 按顺序向所有的mater节点请求加锁。客户端设置网络连接和响应超时时间，并且超时时间要小于锁的失效时间。（假设锁自动失效时间为10秒，则超时时间一般在5-50毫秒之间）。如果超时，跳过该master节点，尽快取尝试下一个master节点。
• 客户端使用当前时间减去开始获取锁时间，得到获取锁使用的时间。当且仅当超过一半的Redis master节点都获得锁，并且使用的时间(所有节点获取锁所用时间与超时时间和)小于锁失效时间时，锁才算获取成功。
• 如果获取到了锁，key的真正有效时间就变了，需要减去获取锁所使用的时间。
• 如果获取锁失败，需要在所有的master节点上解锁。

伪代码如下：

public void testLock(String key) {
    RedissonRedLock redLock = null;
    try {
        // 读取配置文件
        Config config1 = Config.fromYAML(new File("redis-file1.yaml"));
        RedissonClient redissonClient1 = Redisson.create(config1);
        RLock lock1 = redissonClient1.getLock(key);
        Config config2 = Config.fromYAML(new File("redis-file2.yaml"));
        RedissonClient redissonClient2 = Redisson.create(config2);
        RLock lock2 = redissonClient2.getLock(key);
        Config config3 = Config.fromYAML(new File("redis-file3.yaml"));
        RedissonClient redissonClient3 = Redisson.create(config3);
        RLock lock3 = redissonClient3.getLock(key);
        redLock = new RedissonRedLock(lock1, lock2, lock3);
        // 不设置锁的过期时间，启动watchDog机制
        boolean b = redLock.tryLock(10, TimeUnit.SECONDS);
        if (b) {
            // 加锁成功后的业务逻辑
        }
    } catch (Exception e) {
        log.error("加锁失败");
    } finally {
        if (Objects.nonNull(redLock)) {
            redLock.unlock();
        }
    }
}

本栏目推荐文章
• redis 数据库
• 分布式限流——基于Redis的Lua脚本限流实现
• Redis 哨兵启动 以及 手动切换节点
• redis 浅谈3
• Linux 部署redis集群（三主三从）
• 分布式代码执行分析
• Redis持久化之RDB和AOF
• .net core分布式定时任务调度系统
• 1.12_redis 的存取在最后 晚上_浙江本地环境的header不能用线上的_header中host和refer分别代表什么意思?_模型的save()参数是数组怎么理解?
• redis报错MISCONF Errors writing to the AOF file: No space left on device，磁盘满无法写入数据报错

分布式 Redis分布式redis redis答卷 分布式 缓存 利弊 分布式 代码redis 分布式 脚本redis lua 分布式 指南redis 分布式 方法redis 分布式 方案redis 分布式redlock redis core 分布式 系统redis 分布式 建议 金融redis