目录

一、高效分布式锁

二、Redis分布式锁的缺点

三、使用Jedis set命令以及Lua脚本方式实现分布式锁

四、使用Redisson+RLock实现分布式锁

         锁互斥机制 

watch dog自动延期机制

可重入加锁机制

锁释放机制

相关实现

         优缺点

五、参考

---

一、高效分布式锁

1、互斥

在分布式高并发的条件下，我们最需要保证，同一时刻只能有一个线程获得锁，这是最基本的一点。

2、防止死锁

在分布式高并发的条件下，比如有个线程获得锁的同时，还没有来得及去释放锁，就因为系统故障或者其它原因使它无法执行释放锁的命令,导致其它线程都无法获得锁，造成死锁。

所以分布式非常有必要设置锁的有效时间，确保系统出现故障后，在一定时间内能够主动去释放锁，避免造成死锁的情况。

3、性能

对于访问量大的共享资源，需要考虑减少锁等待的时间，避免导致大量线程阻塞。

所以在锁的设计时，需要考虑两点。

1）锁的颗粒度要尽量小。比如你要通过锁来减库存，那这个锁的名称你可以设置成是商品的ID,而不是任取名称。这样这个锁只对当前商品有效,锁的颗粒度小。

2）锁的范围尽量要小。比如只要锁2行代码就可以解决问题的，那就不要去锁10行代码了。

4、重入

我们知道ReentrantLock是可重入锁，那它的特点就是：同一个线程可以重复拿到同一个资源的锁。重入锁非常有利于资源的高效利用。

二、Redis分布式锁的缺点

Redis分布式锁会有个缺陷，就是在Redis哨兵模式下:

客户端1 对某个master节点写入了redisson锁，此时会异步复制给对应的 slave节点。但是这个过程中一旦发生 master节点宕机，主备切换，slave节点从变为了 master节点。

这时客户端2 来尝试加锁的时候，在新的master节点上也能加锁，此时就会导致多个客户端对同一个分布式锁完成了加锁。

这时系统在业务语义上一定会出现问题，导致各种脏数据的产生。

缺陷在哨兵模式或者主从模式下，如果 master实例宕机的时候，可能导致多个客户端同时完成加锁。

三、使用Jedis set命令以及Lua脚本方式实现分布式锁

依赖

<dependency><groupId>redis.clients</groupId><artifactId>jedis</artifactId><version>2.9.0</version>
</dependency>

分布式锁接口类

public interface DistributedLockInterface {boolean tryLock(String lockKey,String lockValue,long expiryTime);boolean unLock(String lockKey,String lockValue);}

实现类

import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import redis.clients.jedis.JedisCluster;
import redis.clients.jedis.exceptions.JedisNoScriptException;import javax.annotation.Resource;
import java.util.ArrayList;
import java.util.List;/*** redis分布式锁**/
@Component
@Slf4j
public class RedisDistributedLock implements DistributedLockInterface {@Resource(name = "jedisCluster")private JedisCluster jedisCluster;private static final String LOCK_SUCCESS_CODE = "ok";private static final Long UNLOCK_SUCCESS_CODE = 1L;private static final String DISTRIBUTE_LOCK_SCRIPT_UNLOCK_VAL = "if" +" redis.call('get', KEYS[1]) == ARGV[1]" +" then" +" return redis.call('del', KEYS[1])" +" else" +" return 0" +" end";private volatile String unlockSha1 = "";@Overridepublic boolean tryLock(String lockKey, String lockValue, long expiryTime) {return tryLock(lockKey, lockValue, expiryTime, 3, 1000);}private boolean tryLock(String lockKey, String lockVal, long expiryTime, int retryTimes, long setpTime) {while (retryTimes > 0) {if (redisLock(lockKey, lockVal, expiryTime)) {return true;}retryTimes--;try {Thread.sleep(setpTime);} catch (InterruptedException e) {log.error("get distribute lock error", e);}}return false;}private boolean redisLock(String lockKey, String lockVal, long expiryTime) {String result = jedisCluster.set(lockKey, lockVal, "NX", "PX", expiryTime);return LOCK_SUCCESS_CODE.equalsIgnoreCase(result);}@Overridepublic boolean unLock(String lockKey, String lockValue) {List<String> keys = new ArrayList<>();keys.add(lockKey);List<String> argv = new ArrayList<>();argv.add(lockValue);try {Object result = jedisCluster.evalsha(unlockSha1, keys, argv);return UNLOCK_SUCCESS_CODE.equals(result);} catch (JedisNoScriptException e) {log.info("try to store script");storeScript(lockKey);Object result = jedisCluster.evalsha(unlockSha1, keys, argv);return UNLOCK_SUCCESS_CODE.equals(result);} catch (Exception e) {e.printStackTrace();return false;}}private void storeScript(String slotKey) {if (StringUtils.isEmpty(unlockSha1) || !jedisCluster.scriptExists(unlockSha1, slotKey)) {unlockSha1 = jedisCluster.scriptLoad(DISTRIBUTE_LOCK_SCRIPT_UNLOCK_VAL, slotKey);}}
}

调用

public boolean test() {boolean result = false;String lockValue = System.currentTimeMillis() + "";try {if (redisDistributedLock.tryLock(LOCK_KEY, lockValue, 5 * 60 * 1000)) {// 业务}} catch (Exception e) {log.warn("test error, errMsg={}", e.getMessage(), e);} finally {redisDistributedLock.unLock(LOCK_KEY, lockValue);}return result;}

缺陷：

1）使用set方法获取分布式锁，没有实现等待时，需要使用者自己实现这个等待时重新获取锁的逻辑；

2）没有实现锁续期功能，一旦过期就释放锁，不管代码是否执行完；

四、使用Redisson+RLock实现分布式锁

加锁调用 tryLockInnerAsync()方法

 加锁流程

解锁调用的是 unlockInnerAsync() 方法

  解锁流程

 

锁互斥机制

如果客户端2来尝试加锁，执行了同样的一段lua脚本。
1.第一个if判断这个key发现已存在，走第二个if判断key中的客户端的id是否与是客户端2的id
2.如果发现不是，客户端2会获取到pttl myLock返回的一个数字，这个数字代表了myLock这个锁key的剩余生存时间。比如还剩15000毫秒的生存时间。此时客户端2会进入一个while循环，不停的尝试加锁。
实际上的原理：
当锁正在被占用时，等待获取锁的进程并不是通过一个 while(true) 死循环去获取锁，而是利用了 Redis 的发布订阅机制,通过 await 方法阻塞等待锁的进程，有效的解决了无效的锁申请浪费资源的问题。

watch dog自动延期机制

客户端1加锁的锁key默认生存时间才30秒，如果超过了30秒，客户端1还想一直持有这把锁，怎么办呢？
简单！只要客户端1一旦加锁成功，就会启动一个watch dog看门狗，他是一个后台线程，会每隔10秒检查一下，如果客户端1还持有锁key，过期时间继续延长为30s。

可重入加锁机制

上面那段lua脚本。第一个if判断肯定不成立，“exists myLock”会显示锁key已经存在了。第二个if判断会成立，因为myLock的hash数据结构中包含的那个ID，就是客户端1的那个ID，也就是“8743c9c0-0795-4907-87fd-6c719a6b4586:1”，此时就会执行可重入加锁的逻辑，他会用：incrby myLock 8743c9c0-0795-4907-87fd-6c71a6b4586:1 1 ，通过这个命令，对客户端1的加锁次数，累加1，变为myLock 8743c9c0-0795-4907-87fd-6c71a6b4586:1 2

锁释放机制

如果执行lock.unlock()，就可以释放分布式锁，此时的业务逻辑也是非常简单的。其实说白了，就是每次都对myLock数据结构中的那个加锁次数减1。如果发现加锁次数是0了，说明这个客户端已经不再持有锁了，此时就会用：“del myLock”命令，从redis里删除这个key

相关实现

依赖

<dependency><groupId>org.redisson</groupId><artifactId>redisson</artifactId><version>3.12.0</version>
</dependency>

配置参数（yml）:

#集群配置
clusterServersConfig:
#  连接空闲超时，单位：毫秒，默认值 10000idleConnectionTimeout: 10000
#  ping节点超时，单位：毫秒，默认值 1000pingTimeout: 1000
#  连接超时，单位：毫秒，默认值 10000connectTimeout: 10000
#  命令等待超时，单位：毫秒，默认值 3000timeout: 3000
#  命令失败重试次数，默认值 3retryAttempts: 3
#  命令重试发送时间间隔，单位：毫秒，默认值 1500retryInterval: 1500
#  重新连接时间间隔，单位：毫秒，默认值 3000reconnectionTimeout: 3000
#  执行失败最大次数，默认值 3failedAttempts: 3
#  密码password: null
#  单个连接最大订阅数量，默认值 5subscriptionsPerConnection: 5
#  客户端名称clientName: null
#  负载均衡算法loadBalancer: !<org.redisson.connection.balancer.RoundRobinLoadBalancer> {}
#  从节点发布和订阅连接的最小空闲连接数，默认值 1slaveSubscriptionConnectionMinimumIdleSize: 1
#  从节点发布和订阅连接池大小，默认值 50slaveSubscriptionConnectionPoolSize: 50
#  从节点最小空闲连接数，默认值 32slaveConnectionMinimumIdleSize: 32
#  从节点连接池大小，默认值 64slaveConnectionPoolSize: 64
#  主节点最小空闲连接数，默认值 32masterConnectionMinimumIdleSize: 32
#  主节点连接池大小，默认值 64masterConnectionPoolSize: 64
#  读取操作的负载均衡模式，默认值 SLAVE（只在从节点里读取）readMode: "SLAVE"
#  节点地址nodeAddresses:- "redis://*.*.*.*:****"
#  集群扫描间隔时间，默认值 1000scanInterval: 1000
# 线程池数量，默认值: 当前处理核数量 * 2
#threads: 0
# Netty线程池数量，默认值: 当前处理核数量 * 2
#nettyThreads: 0
# 编码，默认值 org.redisson.codec.JsonJacksonCodec
#codec: !<org.redisson.codec.JsonJacksonCodec> {}
# 传输模式，默认值 NIO
#"transportMode":"NIO"# 看门狗默认过期时间 默认值 30000 单位：毫秒
lockWatchdogTimeout: 30000

把ReddissonClient作为单例类HTRedissonClient的一个属性，经过过单例类HTRedissonClient初始化，完成ReddissonClient的配置，然后只需要通过调用HTRedissonClient的getRedissonClient()方法，就可以获得ReddissonClient对象进行相应操作
 

public class HTRedissonClient {protected final Logger cLogger = Logger.getLogger(getClass());private static Config config = new Config();private static RedissonClient redissonClient;private static volatile HTRedissonClient instance;private static String cPath="redisson-config.yml";private HTRedissonClient(){init();}private void init() {try {cLogger.info("加载redisson配置文件");//         String filePath= SysInfo.cHome+cPath;String filePath="classpath:redisson-config.yml";cLogger.info("配置文件地址："+filePath);File file = ResourceUtils.getFile(filePath);config = Config.fromYAML(file);cLogger.info("配置成功");cLogger.info("创建RedissonClient实例");redissonClient= Redisson.create(config);cLogger.info("RedissonClient实例创建完成");}catch (Exception e){e.printStackTrace();}}public static HTRedissonClient getInstance(){if(instance==null){synchronized ((HTRedissonClient.class)){if(instance==null){instance = new HTRedissonClient();}}}return instance;}public RedissonClient getRedissonClient(){return redissonClient;}}

使用方式示例，这里要注意一下，Redisson是提供了自动续期的功能的，但是自动续期功能只有在我们不自定义过期时间时才启用；（不自定义过期时间时，默认是30秒过期，启用自动续期功能，默认的过期时间可以通过配置文件参数lockWatchdogTimeout配置）

@Overridepublic void run() {String lockKey="lockKey";RLock rLock = redissonClient.getLock(lockKey);try {//尝试加锁，最多等待5秒，默认30秒过期时间，有自动续期功能if (rLock.tryLock(5, TimeUnit.SECONDS)) {cLogger.info(Thread.currentThread().getName()+"获取锁成功");try {//（业务代码）Thread.sleep(2000);//模拟业务代码执行时间}finally {rLock.unlock();// 释放锁cLogger.info(Thread.currentThread().getName()+"释放锁成功");}}else{cLogger.info(Thread.currentThread().getName()+"未获取到锁");}}catch (Exception e){e.printStackTrace();}}

测试demo

public static void main(String args[]) throws InterruptedException {//获取HTRedissonClient实例对象RedissonClient redissonClient = HTRedissonClient.getInstance().getRedissonClient();for(int i=0;i<10;i++) {//作为入参传入实现了Runnable接口的RedissonTest的构造函数，获得RedissonTesst对象RedissonTest redissonTest=new RedissonTest(redissonClient);//创建线程Thread thread = new Thread(redissonTest);//启动线程thread.start();}}

优缺点

优点：

1、Redisson把获取锁+设置等待时间+设置过期时间、释放锁的各种操作封装成原子操作。

2、通过 Watch Dog 机制很好的解决了锁的续期问题

3、提供了可重入锁的功能

4、在等待申请锁资源的进程等待申请锁的实现上也做了一些优化，减少了无效的锁申请，提升了资源的利用率。

5、和 Zookeeper 相比较，Redisson 基于 Redis 性能更高，适合对性能要求高的场景

缺陷：

unLock（）方法可能会尝试释放别的线程的锁，虽然不会成功，但是会报异常；

五、参考

说一说redis分布式锁的几种实现及优缺点_weixin_43368125的博客-CSDN博客_redis实现分布式锁的缺点

Redisson实现分布式锁(1)---原理 - 雨点的名字 - 博客园

https://segmentfault.com/a/1190000039728618