单体锁存在的问题

在单体应用中，如果我们对共享数据不进行加锁操作，多线程操作共享数据时会出现数据一致性问题。

(下述实例是一个简单的下单问题：从redis中获取库存，检查库存是否够，>0才允许下单)

我们的解决办法通常是加锁。如下加单体锁（synchronized或RentranLock）来保证单个实例并发安全：

但上锁代码块内线程只能串行执行，效率低。单体应用难以满足实际高并发访问需求，会将单体应用部署到多个tomcat实例上，由负载均衡将请求分发到不同实例上。

一个tomocat实例是一个JVM进程，单体锁（synchronized、ReentrantLock）是JVM层面的锁，只能控制单个实例上的并发访问安全，多实例下依然存在数据一致性问题。

分布式锁

分布式锁登场：分布式锁指的是，所有服务中的所有线程都去获取同一把锁，但只有一个线程可以成功的获得锁，其他没有获得锁的线程必须全部等待，直到持有锁的线程释放锁。

分布式锁是可以跨越多个实例，多个进程的锁

分布式锁具备的条件

互斥性：任意时刻，只能有一个客户端持有锁

锁超时释放：持有锁超时，可以释放，防止死锁

可重入性：一个线程获取了锁之后，可以再次对其请求加锁

高可用、高性能：加锁和解锁开销要尽可能低，同时保证高可用

安全性：锁只能被持有该锁的服务(或应用)释放。

容错性：在持有锁的服务崩溃时，锁仍能得到释放，避免死锁。

分布式锁实现方案

分布式锁都是通过第三方组件来实现的，目前比较流行的分布式锁的解决方案有：

1、数据库，通过数据库可以实现分布式锁，但是在高并发的情况下对数据库压力较大，所以很少使用。
2、Redis，借助Redis也可以实现分布式锁，而且Redis的Java客户端种类很多，使用的方法也不尽相同。
3、Zookeeper，Zookeeper也可以实现分布式锁，同样Zookeeper也存在多个Java客户端，使用方法也不相同

（数据库、Zookeeper实现可参考：https://blog.csdn.net/poizxc2014/article/details/123963250，这里主要介绍哈redis实现分布式锁）

Redis实现分布式锁

再回到上述获取库存的实例，使用Redis实现并发访问安全。

1）基本方案：Redis提供了setXX指令来实现分布式锁

SETNX

格式: setnx key value
将key 的值设为value ，当且仅当key不存在。
若给定的 key已经存在，则SETNX不做任何动作。

设置分布式锁后，能保证并发安全，但上述代码还存在问题，如果执行过程中出现异常，程序就直接抛出异常退出，导致锁没有释放造成最终死锁的问题。（即使将锁放在finally中释放，但是假如是执行到中途系统宕机，锁还是没有被成功的释放掉，依然会出现死锁现象）

2）方案改进：可以给锁设置一个超时时间，到时自动释放锁（锁的过期时间大于业务执行时间）

上述两行代码中，由于加锁和设置锁过期时间不是原子的，可能加锁完就宕机了，那死锁依然存在，所以需要保证两指令执行的原子性

连起来一起写可以原子执行。

3）改进三：再看看是否还有问题。假设有多个线程，锁的过期时间10s,线程1上锁后执行业务逻辑的时长超过十秒，锁到期释放锁，线程2就可以获得锁执行，此时线程1执行完删除锁,删除的就是线程2持有的锁，线程3又可以获取锁，线程2执行完删除锁，删除的是线程3的锁，如此往后，这样就会出问题。

解决办法就是让线程只能删除自己的锁，即给每个线程上的锁添加唯一标识（这里UUID实现，基本不会出现重复），删除锁时判断这个标识：

但上述红框中由于判定和释放锁不是原子的，极端情况下，可能判定可以释放锁，在执行删除锁操作前刚好时间到了，其他线程获取锁执行，前者线程删除锁删除的依然是别的线程的锁，所以要让删除锁具有原子性，可以利用redis事务或lua脚本实现原子操作判断+删除

//redis事务或lua脚本（lua脚本的执行是原子的）,如下

    @RequestMapping(" /deduct_stock")
    public String deductStock() {
        String REDIS_LOCK = "good_lock";
        // 每个人进来先要进行加锁，key值为"good_lock"
        String value = UUID.randomUUID().toString().replace("-","");
        try{
            // 为key加一个过期时间
            Boolean flag = template.opsForValue().setIfAbsent(REDIS_LOCK, value,10L,TimeUnit.SECONDS);

            // 加锁失败
            if(!flag){
                return "抢锁失败！";
            }
            System.out.println( value+ " 抢锁成功");
            String result = template.opsForValue().get("goods:001");
            int total = result == null ? 0 : Integer.parseInt(result);
            if (total > 0) {
                // 如果在此处需要调用其他微服务，处理时间较长。。。
                int realTotal = total - 1;
                template.opsForValue().set("goods:001", String.valueOf(realTotal));
                System.out.println("购买商品成功，库存还剩：" + realTotal + "件， 服务端口为8002");
                return "购买商品成功，库存还剩：" + realTotal + "件， 服务端口为8002";
            } else {
                System.out.println("购买商品失败，服务端口为8002");
            }
            return "购买商品失败，服务端口为8002";
        }finally {
            // 谁加的锁，谁才能删除
            // 也可以使用redis事务
            // https://redis.io/commands/set
            // 使用Lua脚本，进行锁的删除

            Jedis jedis = null;
            try{
                jedis = RedisUtils.getJedis();

                String script = "if redis.call('get',KEYS[1]) == ARGV[1] " +
                        "then " +
                        "return redis.call('del',KEYS[1]) " +
                        "else " +
                        "   return 0 " +
                        "end";

                Object eval = jedis.eval(script, Collections.singletonList(REDIS_LOCK), Collections.singletonList(value));
                if("1".equals(eval.toString())){
                    System.out.println("-----del redis lock ok....");
                }else{
                    System.out.println("-----del redis lock error ....");
                }
            }catch (Exception e){

            }finally {

                if(null != jedis){
                    jedis.close();
                }
            }

            // redis事务
//            while(true){
//                template.watch(REDIS_LOCK);
//                if(template.opsForValue().get(REDIS_LOCK).equalsIgnoreCase(value)){
//                    template.setEnableTransactionSupport(true);
//                    template.multi();
//                    template.delete(REDIS_LOCK);
//                    List<Object> list = template.exec();
//                    if(list == null){
//                        continue;
//                    }
//                }
//                template.unwatch();
//                break;
//            }
        }
        
    }
}

4）当然，也有不错的框架解决该问题，如Redission，Redisson是redis官网推荐实现分布式锁的一个第三方类库，通过开启另一个服务，后台进程定时检查持有锁的线程是否继续持有锁了，是将锁的生命周期重置到指定时间，即防止线程释放锁之前过期，所以将锁声明周期通过重置延长）

如下，先引入依赖，并在在主启动类中加入如下配置：

Redission执行流程如下：（只要线程一加锁成功，就会启动一个watch dog看门狗，它是一个后台线程，会每隔10秒检查一下（锁续命周期就是设置的超时时间的三分之一），如果线程还持有锁，就会不断的延长锁key的生存时间。因此，Redis就是使用Redisson解决了锁过期释放，业务没执行完问题。当业务执行完，释放锁后，再关闭守护线程，

外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-jgMaSCQd-1658890675882)(network-img/image-20220707014528848.png)
其他未获取锁的线程一直自旋。

前面几种方案都只是基于单机版的讨论，但是生产环境中如果是单机服务挂了，缓存就挂了，还不是很完美。为了实现高可用redis会集群部署

但在这种情况下又会出现锁丢失问题。

如果线程一在Redis的master节点上拿到了锁，但是加锁的key还没同步到slave节点。恰好这时，master节点发生故障，一个slave节点就会升级为master节点。线程二就可以获取同个key的锁啦，这就出现线程A还没执行完，线程B又来执行了，就会有并发安全问题。

为了解决这个问题，Redis作者 antirez提出一种高级的分布式锁算法：Redlock。redlock是一种基于多节点redis实现分布式锁的算法，可以有效解决redis单点故障的问题。官方建议搭建五台redis服务器对redlock算法进行实现。

Redlock原理：搞多个Redis master部署，以保证它们不会同时宕掉。并且这些master节点是完全相互独立的，相互之间不存在数据同步。同时，需要确保在这多个master实例上，是与在Redis单实例，使用相同方法来获取和释放锁。当超过半数的redis节点加锁成功才算成功获取锁

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-vo7obWhO-

RedLock的实现步骤:如下

1.获取当前时间，以毫秒为单位。
2.使用相同的key，value按顺序向5个master节点请求加锁，客户端设置网络连接和响应超时时间，并且设置获取锁的时间要远远小于锁自动释放的时间。假设锁自动释放时间是10秒，则获取时间应在5-50毫秒之间。通过这种方式避免客户端长时间等待一个已经关闭的实例，如果一个实例不可用了，则立即尝试获取下一个实例。
3.客户端使用当前时间减去开始获取锁时间（即步骤1记录的时间），得到获取锁使用的时间。使用的时间小于锁失效时间时，避免拿到一个已经过期的锁，并且要有超过半数的redis实例成功获取到锁，才算最终获取锁成功。如果不是超过半数，有可能出现多个客户端重复获取到锁，导致锁失效。（如上图，10s> 30ms+40ms+50ms+4m0s+50ms）
4.

如果取到了锁，key的正失效时间应该为：过期时间-第三步的差值。
如果获取锁失败（没有在至少N/2+1个master实例取到锁，有或者获取锁时间已经超过了有效时间），客户端要在所有的master节点上解锁（即便有些master节点根本就没有加锁成功，也需要解锁，以防止有些漏网之鱼）。

为了保证更高效的获取锁，还可以设置重试策略，在一定时间后重新尝试获取锁，但不能是无休止的，要设置重试次数。
简化下步骤就是：

按顺序向5个master节点请求加锁
根据设置的超时时间来判断，是不是要跳过该master节点。
如果大于等于3个节点加锁成功，并且使用的时间小于锁的有效期，即可认定加锁成功啦。
如果获取锁失败，解锁！
Redisson实现了redLock版本的锁，有兴趣的小伙伴，可以去了解一下哈~

因为已经有半数的redis加锁了，假如有第二个线程来加锁，没加锁的redis少于一半，这样就没法加锁成功，从而保证并发安全

虽然通过redlock能够更加有效的防止redis单点问题，但是仍然是存在隐患的。假设redis没有开启持久化，clientA获取锁后，所有redis故障重启，则会导致clientA锁记录消失，clientB仍然能够获取到锁。这种情况虽然发生几率极低，但并不能保证肯定不会发生。

或者假设redis1,2都加锁成功并返回，redis3加锁成功返回，但是还没来得及同步就挂了，从节点变主节点，此时又有一个线程来加锁，reids3,4,5都set成功，那该线程也获取了锁，这样又会出问题。

保证的方案就是开始AOF持久化，但是要注意同步的策略，使用每秒同步，如果在一秒内重启，仍然数据丢失。使用always又会造成性能急剧下降。

官方推荐使用默认的AOF策略即每秒同步，且在redis停掉后，要在ttl时间后再重启。 缺点就是ttl时间内redis无法对外提供服务。

Redis与zookeeper分布式锁对比

RedLock实现分布式锁方式类似Zookeeper实现的分布式锁，zookeeper节点分布是树形结构，当某线程加锁向zookeeper中写入key后并不会立即返回结果，而是至少半数以上的从节点同步成功key后，才会返回上锁成功的结果，如果主节点挂了，他会确保选举出的从节点key一定存在。和redlock相比，虽然性能比redis实现慢了点，但是能确保系统安全

Redis与zookeeper分布式锁对比

• redis集群是AP,zookeeper集群是CP,redis在集群架构上性能很高，zookeeper在数据一致性上做的更好。

• redis往主节点写成功key，马上告诉客户端写成功，收到半数的返回结果就可以执行代码逻辑。所以redis采用抢占式方式进行锁的获取，需要不断的在用户态进行CAS尝试获取锁，对CPU占用率高。

• zookeeper在主节点中写好key，会把key同步给所有从节点，并且接受从节点是否同步成功返回。当主节点接收到半数以上的同步成功的结果,就会返回客户端可以执行业务逻辑了。

• Zookeeper不会出现主从架构锁丢失问题，主节点挂了，zookeeper的ZAB选举机制一定会把同步成功的从节点变为主节点。

• 如果追求正确，就选zk集群，如果允许一点出错，追求性能，那选redis

• 对于redis分布式锁的使用，在企业中是非常常见的，绝大多数情况不会出现极端情况。

redis实现分布式锁总结：

1)setnx:redis提供的分布式锁
存在问题：线程还没释放锁系统宕机了，造成死锁
2）setnx +setex：给锁设置过期时间，到期自动删除。
存在问题：因为加锁和过期时间设置非原子，存在设置超时时间失败情况，导致死锁
3）set(key,value,nx,px):将setnx+setex变成原子操作
存在问题：加锁和释放锁不是同一个线程的问题。假如线程1业务还没执行完，锁过期释放，线程2获取锁执行，线程1执行完业务删除锁删除的就是线程2的，然后其他线程又可获取锁执行，线程2执行完释放锁删除的是别人的，如此往复，导致并发安全问题。
4.方法1：在value中存入uuid(线程唯一标识)，删除锁时判断该标识，同时删除锁需保证原子性，否则还是有删除别人锁问题，可通过lua或者redis事务释放锁
方法2：利用redis提供的第三方类库，Redisson也可解决任务超时，锁自动释放问题。其通过开启另一个服务，后台进程定时检查持有锁的线程是否继续持有锁了，是将锁的生命周期重置到指定时间，即防止线程释放锁之前过期，所以将锁声明周期通过重置延长。
Redission也可解决不可重入问题（AQS,计数）

问题：但上述方案能保证单机系统下的并发访问安全，实际为了保证redis高可用，redis一般会集群部署。单机解决方案会出现锁丢失问题。如线程set值后成功获取锁但主节点还没来得及同步就宕机了，从节点选举成为主节点，没有锁信息，此时其他线程就可以加锁成功，导致并发问题。

5）redis集群解决方案,使用redlock解决:

• 顺序向5个节点请求加锁（5个节点相互独立，没任何关系）
• 根据超时时间来判断是否要跳过该节点
• 如果大于等于3节点加锁成功，并且使用时间小于锁有效期，则加锁成功，否则获取锁失败，解锁

参考文档：https://blog.csdn.net/poizxc2014/article/details/123963250