第二十章_Redis分布式锁Redlock算法和底层源码分析

当前代码为8.0版接上一步

自研一把分布式锁，面试中回答的主要考点

按照JUC里面java.util.concurrent.locks.Lock接口规范编写

lock()加锁关键逻辑

加锁的Lua脚本，通过redis里面的hash数据模型，加锁和可重入性都要保证

加锁不成，需要while进行重试并自旋

自动续期，加个钟

加锁

加锁实际上就是在redis中，给Key键设置一个值，为避免死锁，并给定一个过期时间

自旋

续期

unlock解锁关键逻辑

考虑可重入性的递减，加锁几次就要减锁几次

最后到零了，直接del删除

将Key键删除。但也不能乱删，不能说客户端1的请求将客户端2的锁给删除掉，只能自己删除自己的锁

上面自研的redis锁对于一般中小公司，不是特别高并发场景足够用了，单机redis小业务也撑得住

Redis分布式锁-Redlock红锁算法Distributed locks with Redis

RedLock是什么？

RedLock是基于redis实现的分布式锁，它能够保证以下特性：

互斥性：在任何时候，只能有一个客户端能够持有锁；

避免死锁：当客户端拿到锁后，即使发生了网络分区或者客户端宕机，也不会发生死锁；（利用key的存活时间）

容错性：只要多数节点的redis实例正常运行，就能够对外提供服务，加锁或者释放锁；

而非redLock是无法满足互斥性的，上面已经阐述过了原因。

RedLock算法

假设有N个redis的master节点，这些节点是相互独立的（不需要主从或者其他协调的系统）。N推荐为奇数~

客户端在获取锁时，需要做以下操作：

获取当前时间戳，以微妙为单位。

使用相同的lockName和lockValue，尝试从N个节点获取锁。（在获取锁时，要求等待获取锁的时间远小于锁的释放时间，如锁的lease_time为10s，那么wait_time应该为5-50毫秒；避免因为redis实例挂掉，客户端需要等待更长的时间才能返回，即需要让客户端能够fast_fail；如果一个redis实例不可用，那么需要继续从下个redis实例获取锁）

当从N个节点获取锁结束后，如果客户端能够从多数节点(N/2 + 1)中成功获取锁，且获取锁的时间小于失效时间，那么可认为，客户端成功获得了锁。（获取锁的时间=当前时间戳 - 步骤1的时间戳）

客户端成功获得锁后，那么锁的实际有效时间 = 设置锁的有效时间 - 获取锁的时间。

客户端获取锁失败后，N个节点的redis实例都会释放锁，即使未能加锁成功。

为什么N推荐为奇数呢？

原因1：本着最大容错的情况下，占用服务资源最少的原则，2N+1和2N+2的容灾能力是一样的，所以采用2N+1；比如，5台服务器允许2台宕机，容错性为2，6台服务器也只能允许2台宕机，容错性也是2，因为要求超过半数节点存活才OK。

原因2：假设有6个redis节点，client1和client2同时向redis实例获取同一个锁资源，那么可能发生的结果是——client1获得了3把锁，client2获得了3把锁，由于都没有超过半数，那么client1和client2获取锁都失败，对于奇数节点是不会存在这个问题。

红锁算法原理

分布式锁假设有N个redis 主节点，这些节点都是独立的，我们不用复本或者其他隐性协调系统，我们假设有5个节点，这是一个有依据的值，因此，我们需要在不同的计算机或虚拟机上运行5个Redis主机，确保他们失败的时候是独立的、加锁时客户端执行以下操作。

1、获取当前毫秒时间。

2、在5个实例上依次获取锁，在所有实例中使用相同的key名字和随机值，在步骤2期间，当设置锁再每个实例时，客户端使用与总锁自动释放时间相比小的超时来获取它。如果锁的自动释放时间是10秒，那么客户端的超时时间是 5-50 毫秒，这可防止客户端在尝试与已关闭的Redis节点通话时长时间处于阻塞状态，如果实例宕机，我们应该尽快尝试连接下一个实例。

3、客户端计算获取锁所需的时间通过从当前时间减去步骤1中获得的时间戳，当且仅当客户端能够在大多数实例（至少3个）中获取锁时，获取锁所花费的总时间小于锁有效时间，则认为已获取锁。

4、如果加锁成功，其有效时间被认为是初始有效时间减去在步骤3中计算的经过时间，就是锁的有效时间 == 10-获取锁花费的时间。

5、如果锁获取失败了，不管是因为获取成功的redis节点没有过半，还是因为获取锁的总耗时超过了锁的超时时间，都会向已经获取锁成功的redis实例发出删除对应key的请求，它将尝试解锁所有实例。

这个算法安全吗？

首先，我们假设客户端能够在大多数实例中获取锁所有实例都将包含一个具有相同生存时间的密钥，然而，这个key是在不用时间设置的，所以过期也会在不同的时间过期，假设第一个key设置时间T1，最后一个key设置时间t2，我们确定第一个key的过期时间至少是MIN_VALIDITY=TTL-(T2-T1)-CLOCK_DRIFT，所以所有的key设置超时至少是这个时间，在设置大多数密钥期间，另一个客户端将无法获取锁，因为如果已经存在N/2+1个密钥，则N/2+1 set NX操作无法成功。因此，如果获取了锁，则不可能同时重新获取它。

如果客户端使用接近或大于锁定最大有效时间（我们基本上用于SET的TTL）的时间锁定了大多数实例，它将认为锁定无效并将解锁实例，因此我们只需要考虑客户端能够在小于有效时间的时间内锁定大多数实例的情况。在这种情况下，对于上面已经表达的参数，对于MIN_VALIDITY，任何客户端都不能重新获取锁。因此，只有当锁定大多数实例的时间大于TTL时间时，多个客户端才能同时锁定N/2+1个实例（“时间”是步骤2的结尾），从而使锁定无效。

官网说明

redis分布式锁官网

主页说明

为什么学习这个？怎么产生的？

主从redis架构中分布式锁存在的问题

1、线程A从主redis中请求一个分布式锁，获取锁成功；

2、从redis准备从主redis同步锁相关信息时，主redis突然发生宕机，锁丢失了；

3、触发从redis升级为新的主redis；

4、线程B从继任主redis的从redis上申请一个分布式锁，此时也能获取锁成功；

5、导致，同一个分布式锁，被两个客户端同时获取，没有保证独占使用特性；

为了解决这个问题，redis引入了红锁的概念。

之前我们手写的分布式锁有什么缺点？

官网证据

翻译

简单说

线程 1 首先获取锁成功，将键值对写入 redis 的 master 节点，在 redis 将该键值对同步到 slave 节点之前，master 发生了故障；redis 触发故障转移，其中一个 slave 升级为新的 master，此时新上位的master并不包含线程1写入的键值对，因此线程 2 尝试获取锁也可以成功拿到锁，此时相当于有两个线程获取到了锁，可能会导致各种预期之外的情况发生，例如最常见的脏数据。

我们加的是排它独占锁，同一时间只能有一个建redis锁成功并持有锁，严禁出现2个以上的请求线程拿到锁。危险的。

Redlock算法设计理念

redis之父提出了Redlock算法解决这个问题

Redis也提供了Redlock算法，用来实现基于多个实例的分布式锁。

锁变量由多个实例维护，即使有实例发生了故障，锁变量仍然是存在的，客户端还是可以完成锁操作。

Redlock算法是实现高可靠分布式锁的一种有效解决方案，可以在实际开发中使用。最下方还有笔记。

设计理念

该方案也是基于（set 加锁、Lua 脚本解锁）进行改良的，所以redis之父antirez 只描述了差异的地方，大致方案如下。

假设我们有N个Redis主节点，例如 N = 5这些节点是完全独立的，我们不使用复制或任何其他隐式协调系统，

为了取到锁客户端执行以下操作：

1	获取当前时间，以毫秒为单位；
2	依次尝试从5个实例，使用相同的 key 和随机值（例如 UUID）获取锁。当向Redis 请求获取锁时，客户端应该设置一个超时时间，这个超时时间应该小于锁的失效时间。例如你的锁自动失效时间为 10 秒，则超时时间应该在 5-50 毫秒之间。这样可以防止客户端在试图与一个宕机的 Redis 节点对话时长时间处于阻塞状态。如果一个实例不可用，客户端应该尽快尝试去另外一个 Redis 实例请求获取锁；
3	客户端通过当前时间减去步骤 1 记录的时间来计算获取锁使用的时间。当且仅当从大多数（N/2+1，这里是 3 个节点）的 Redis 节点都取到锁，并且获取锁使用的时间小于锁失效时间时，锁才算获取成功；
4	如果取到了锁，其真正有效时间等于初始有效时间减去获取锁所使用的时间（步骤 3 计算的结果）。
5	如果由于某些原因未能获得锁（无法在至少 N/2 + 1 个 Redis 实例获取锁、或获取锁的时间超过了有效时间），客户端应该在所有的 Redis 实例上进行解锁（即便某些Redis实例根本就没有加锁成功，防止某些节点获取到锁但是客户端没有得到响应而导致接下来的一段时间不能被重新获取锁）。

该方案为了解决数据不一致的问题，直接舍弃了异步复制只使用 master 节点，同时由于舍弃了 slave，为了保证可用性，引入了 N 个节点，官方建议是 5。本次演示用3台实例来做说明。

客户端只有在满足下面的这两个条件时，才能认为是加锁成功。

条件1：客户端从超过半数（大于等于N/2+1）的Redis实例上成功获取到了锁；

条件2：客户端获取锁的总耗时没有超过锁的有效时间。

解决方案

为什么是奇数？ N = 2X + 1 (N是最终部署机器数，X是容错机器数)

1 先知道什么是容错

失败了多少个机器实例后我还是可以容忍的，所谓的容忍就是数据一致性还是可以Ok的，CP数据一致性还是可以满足

加入在集群环境中，redis失败1台，可接受。2X+1 = 2 * 1+1 =3，部署3台，死了1个剩下2个可以正常工作，那就部署3台。

加入在集群环境中，redis失败2台，可接受。2X+1 = 2 * 2+1 =5，部署5台，死了2个剩下3个可以正常工作，那就部署5台。

2 为什么是奇数？

最少的机器，最多的产出效果

加入在集群环境中，redis失败1台，可接受。2N+2= 2 * 1+2 =4，部署4台

加入在集群环境中，redis失败2台，可接受。2N+2 = 2 * 2+2 =6，部署6台

容错公式

天上飞的理念(RedLock)必然有落地的实现(Redisson)

redisson实现

Redisson是java的redis客户端之一，提供了一些api方便操作redis

redisson官网

redissonGithub

redisson解决分布式锁

使用Redisson进行编码改造V9.0

你怎么知道该这样使用？

官网说话

分布式锁和同步器

V9.0版本修改

POM添加redisson依赖

<!--redisson-->
<dependency>
    <groupId>org.redisson</groupId>
    <artifactId>redisson</artifactId>
    <version>3.13.4</version>
</dependency>

RedisConfig

package com.test.redislock.config;

import org.redisson.Redisson;
import org.redisson.api.RedissonClient;
import org.redisson.config.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.serializer.GenericJackson2JsonRedisSerializer;
import org.springframework.data.redis.serializer.StringRedisSerializer;

/**
 * @auther admin
 */
@Configuration
public class RedisConfig {

    @Bean
    public RedisTemplate<String, Object> redisTemplate(LettuceConnectionFactory lettuceConnectionFactory) {
        RedisTemplate<String,Object> redisTemplate = new RedisTemplate<>();
        redisTemplate.setConnectionFactory(lettuceConnectionFactory);
        //设置key序列化方式string
        redisTemplate.setKeySerializer(new StringRedisSerializer());
        //设置value的序列化方式json
        redisTemplate.setValueSerializer(new GenericJackson2JsonRedisSerializer());

        redisTemplate.setHashKeySerializer(new StringRedisSerializer());
        redisTemplate.setHashValueSerializer(new GenericJackson2JsonRedisSerializer());

        redisTemplate.afterPropertiesSet();

        return redisTemplate;
    }

    //单Redis节点模式
    @Bean
    public Redisson redisson() {
        Config config = new Config();
        config.useSingleServer().setAddress("redis://192.168.111.175:6379").setDatabase(0).setPassword("111111");
        return (Redisson) Redisson.create(config);
    }

}

InventoryController

package com.test.redislock.controller;

import com.test.redislock.service.InventoryService;
import io.swagger.annotations.Api;
import io.swagger.annotations.ApiOperation;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;

/**
 * @auther admin
 */
@RestController
@Api(tags = "redis分布式锁测试")
public class InventoryController {

    @Autowired
    private InventoryService inventoryService;

    @ApiOperation("扣减库存，一次卖一个")
    @GetMapping(value = "/inventory/sale")
    public String sale() {
        return inventoryService.sale();
    }

    @ApiOperation("扣减库存saleByRedisson，一次卖一个")
    @GetMapping(value = "/inventory/saleByRedisson")
    public String saleByRedisson() {
        return inventoryService.saleByRedisson();
    }

}

从现在开始不再用我们自己手写的锁了

InventoryService

package com.test.redislock.service;
 
/**
 * @auther admin
 */
public class InventoryService {
 
    /**
     * 模拟商品库存扣减
     */
    String sale();

    /**
     * 模拟商品库存扣减(redisson)
     */
    String saleByRedisson()
 
}

InventoryServiceImpl

package com.atguigu.redislock.service.impl;
 
import com.test.redislock.service.InventoryService;
 
import cn.hutool.core.util.IdUtil;
import com.atguigu.redislock.mylock.DistributedLockFactory;
import org.redisson.Redisson;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.stereotype.Service;
 
import java.util.Arrays;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
 
/**
 * @auther admin
 */
@Slf4j
@Service
public class InventoryServiceImpl implements InventoryService {
 
    @Autowired
    private StringRedisTemplate stringRedisTemplate;
 
    @Value("${server.port}")
    private String port;
 
    @Autowired
    private DistributedLockFactory distributedLockFactory;

    @Autowired
    private Redisson redisson;
 
    /**
     * 模拟商品库存扣减
     */
    @Override
    public String sale() {
        String retMessage = "";
        Lock redisLock = distributedLockFactory.getDistributedLock("redis");
        redisLock.lock();
 
        try {
            //1 查询库存信息
            String result = stringRedisTemplate.opsForValue().get("inventory001");
            //2 判断库存是否足够
            Integer inventoryNumber = result == null ? 0 : Integer.parseInt(result);
 
            //3 扣减库存
            if (inventoryNumber > 0) {
                stringRedisTemplate.opsForValue().set("inventory001", String.valueOf(--inventoryNumber));
                retMessage = "成功卖出一个商品，库存剩余: " + inventoryNumber;
                System.out.println(retMessage);
 
                //暂停几秒钟线程,为了测试自动续期
                try { 
                    TimeUnit.SECONDS.sleep(120);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            } else {
                retMessage = "商品卖完了，o(╥﹏╥)o";
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            redisLock.unlock();
        }
 
        return retMessage + "\t" + "服务端口号：" + port;
    }
 
    /**
     * 模拟商品库存扣减(redisson)
     */
    @Override
    public String saleByRedisson() {
        String retMessage = "";
        String key = "zzyyRedisLock";
        RLock redissonLock = redisson.getLock(key);
        redissonLock.lock();

        try {
            //1 查询库存信息
            String result = stringRedisTemplate.opsForValue().get("inventory001");
            //2 判断库存是否足够
            Integer inventoryNumber = result == null ? 0 : Integer.parseInt(result);

            //3 扣减库存
            if(inventoryNumber > 0) {
                stringRedisTemplate.opsForValue().set("inventory001", String.valueOf(--inventoryNumber));
                retMessage = "成功卖出一个商品，库存剩余: " + inventoryNumber;
                System.out.println(retMessage);
            } else {
                retMessage = "商品卖完了，o(╥﹏╥)o";
            }
        } finally {
            redissonLock.unlock();
        }

        return retMessage + "\t" + "服务端口号：" + port;
    }

    /**
     * 可重入性测试方法
     */
    private void testReEnter() {
        Lock redisLock = distributedLockFactory.getDistributedLock("redis");
        redisLock.lock();
 
        try {
            System.out.println("################测试可重入锁####################################");
        } finally {
            redisLock.unlock();
        }
    }
 
}

测试

单机（OK）

JMeter

Bug

解决

业务代码修改为V9.1版

InventoryServiceImpl

package com.atguigu.redislock.service.impl;
 
import com.test.redislock.service.InventoryService;
 
import cn.hutool.core.util.IdUtil;
import com.atguigu.redislock.mylock.DistributedLockFactory;
import org.redisson.Redisson;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.stereotype.Service;
 
import java.util.Arrays;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
 
/**
 * @auther admin
 */
@Slf4j
@Service
public class InventoryServiceImpl implements InventoryService {
 
    @Autowired
    private StringRedisTemplate stringRedisTemplate;
 
    @Value("${server.port}")
    private String port;
 
    @Autowired
    private DistributedLockFactory distributedLockFactory;

    @Autowired
    private Redisson redisson;
 
    /**
     * 模拟商品库存扣减
     */
    @Override
    public String sale() {
        String retMessage = "";
        Lock redisLock = distributedLockFactory.getDistributedLock("redis");
        redisLock.lock();
 
        try {
            //1 查询库存信息
            String result = stringRedisTemplate.opsForValue().get("inventory001");
            //2 判断库存是否足够
            Integer inventoryNumber = result == null ? 0 : Integer.parseInt(result);
 
            //3 扣减库存
            if (inventoryNumber > 0) {
                stringRedisTemplate.opsForValue().set("inventory001", String.valueOf(--inventoryNumber));
                retMessage = "成功卖出一个商品，库存剩余: " + inventoryNumber;
                System.out.println(retMessage);
 
                //暂停几秒钟线程,为了测试自动续期
                try { 
                    TimeUnit.SECONDS.sleep(120);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            } else {
                retMessage = "商品卖完了，o(╥﹏╥)o";
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            redisLock.unlock();
        }
 
        return retMessage + "\t" + "服务端口号：" + port;
    }
 
    /**
     * 模拟商品库存扣减(redisson)
     */
    @Override
    public String saleByRedisson() {
        String retMessage = "";
        String key = "zzyyRedisLock";
        RLock redissonLock = redisson.getLock(key);
        redissonLock.lock();

        try {
            //1 查询库存信息
            String result = stringRedisTemplate.opsForValue().get("inventory001");
            //2 判断库存是否足够
            Integer inventoryNumber = result == null ? 0 : Integer.parseInt(result);

            //3 扣减库存
            if(inventoryNumber > 0) {
                stringRedisTemplate.opsForValue().set("inventory001", String.valueOf(--inventoryNumber));
                retMessage = "成功卖出一个商品，库存剩余: " + inventoryNumber;
                System.out.println(retMessage);
            } else {
                retMessage = "商品卖完了，o(╥﹏╥)o";
            }
        } finally {
            if(redissonLock.isLocked() && redissonLock.isHeldByCurrentThread()) {
                redissonLock.unlock();
            }
        }

        return retMessage + "\t" + "服务端口号：" + port;
    }

    /**
     * 可重入性测试方法
     */
    private void testReEnter() {
        Lock redisLock = distributedLockFactory.getDistributedLock("redis");
        redisLock.lock();
 
        try {
            System.out.println("################测试可重入锁####################################");
        } finally {
            redisLock.unlock();
        }
    }
 
}

Redisson源码解析

加锁
可重入
续命
解锁

分析步骤

Redis 分布式锁过期了，但是业务逻辑还没处理完怎么办

还记得之前说过的缓存续命吗？

守护线程“续命”

额外起一个线程，定期检查线程是否还持有锁，如果有则延长过期时间。

Redisson 里面就实现了这个方案，使用“看门狗”定期检查（每1/3的锁时间检查1次），如果线程还持有锁，则刷新过期时间；

在获取锁成功后，给锁加一个 watchdog，watchdog会起一个定时任务，在锁没有被释放且快要过期的时候会续期。

上述源码分析1

通过redisson新建出来的锁key，默认是30秒

上述源码分析2

RedissonLock.java

lock()---tryAcquire()---tryAcquireAsync()---

上述源码分析3

流程解释

通过exists判断，如果锁不存在，则设置值和过期时间，加锁成功。

通过hexists判断，如果锁已存在，并且锁的是当前线程，则证明是重入锁，加锁成功。

如果锁已存在，但锁的不是当前线程，则证明有其他线程持有锁。返回当前锁的过期时间(代表了锁key的剩余生存时间)，加锁失败。

上述源码分析4

这里面初始化了一个定时器，dely 的时间是 internalLockLeaseTime/3。

在 Redisson 中，internalLockLeaseTime 是 30s，也就是每隔 10s 续期一次，每次 30s。

watch dog自动延期机制

客户端A加锁成功，就会启动一个watch dog看门狗，他是一个后台线程，会每隔10秒检查一下，如果客户端A还持有锁key，那么就会不断的延长锁key的生存时间，默认每次续命又从30秒新开始

自动续期lua脚本分析

解锁

源码分析5

在Redisson框架中，实现了红锁的机制，Redisson的RedissonRedLock对象实现了Redlock介绍的加锁算法。该对象也可以用来将多个RLock对象关联为一个红锁，每个RLock对象实例可以来自于不同的Redisson实例。当红锁中超过半数的RLock加锁成功后，才会认为加锁是成功的，这就提高了分布式锁的高可用。

红锁加锁流程

RedissonRedLock红锁继承自RedissonMultiLock联锁，简单介绍一下联锁：

基于Redis的Redisson分布式联锁RedissonMultiLock对象可以将多个RLock对象关联为一个联锁，每个RLock对象实例可以来自于不同的Redisson实例，所有的锁都上锁成功才算成功。

RedissonRedLock的加锁、解锁代码都是使用RedissonMultiLock中的方法，只是其重写了一些方法，如：

failedLocksLimit()：允许加锁失败节点个数限制。在RedissonRedLock中，必须超过半数加锁成功才能算成功，其实现为：

protected int failedLocksLimit() {
    return locks.size() - minLocksAmount(locks);
}

protected int minLocksAmount(final List<RLock> locks) {
    // 最小的获取锁成功数：n/2 + 1。 过半机制
    return locks.size()/2 + 1;
}

在RedissonMultiLock中，则必须全部都加锁成功才算成功，所以允许加锁失败节点个数为0，其实现为：

protected int failedLocksLimit() {
    return 0;
}

接下来，以tryLock()方法为例，详细分析红锁是如何加锁的，具体代码如下：

org.redisson.RedissonMultiLock#tryLock(long, long, java.util.concurrent.TimeUnit)

public boolean tryLock(long waitTime, long leaseTime, TimeUnit unit) throws InterruptedException {
//        try {
//            return tryLockAsync(waitTime, leaseTime, unit).get();
//        } catch (ExecutionException e) {
//            throw new IllegalStateException(e);
//        }
    long newLeaseTime = -1;
    if (leaseTime > 0) {
        if (waitTime > 0) {
            newLeaseTime = unit.toMillis(waitTime)*2;
        } else {
            newLeaseTime = unit.toMillis(leaseTime);
        }
    }
    // 获取当前系统时间，单位：毫秒
    long time = System.currentTimeMillis();
    long remainTime = -1;
    if (waitTime > 0) {
        remainTime = unit.toMillis(waitTime);
    }
    long lockWaitTime = calcLockWaitTime(remainTime);
    // 允许加锁失败节点个数限制（N - ( N / 2 + 1 )）
    // 假设有三个redis节点，则failedLocksLimit = 1
    int failedLocksLimit = failedLocksLimit();
    // 存放调用tryLock()方法加锁成功的那些redis节点
    List<RLock> acquiredLocks = new ArrayList<>(locks.size());
    // 循环所有节点，通过EVAL命令执行LUA脚本进行加锁
    for (ListIterator<RLock> iterator = locks.listIterator(); iterator.hasNext();) {
        // 获取到其中一个redis实例
        RLock lock = iterator.next();
        String lockName = lock.getName();
        System.out.println("lockName = " + lockName + "正在尝试加锁...");
        boolean lockAcquired;
        try {
            // 未指定锁超时时间和获取锁等待时间的情况
            if (waitTime <= 0 && leaseTime <= 0) {
                // 调用tryLock()尝试加锁
                lockAcquired = lock.tryLock();
            } else {
                // 指定了超时时间的情况,重新计算获取锁的等待时间
                long awaitTime = Math.min(lockWaitTime, remainTime);
                // 调用tryLock()尝试加锁
                lockAcquired = lock.tryLock(awaitTime, newLeaseTime, TimeUnit.MILLISECONDS);
            }
        } catch (RedisResponseTimeoutException e) {
            // 如果抛出RedisResponseTimeoutException异常，为了防止加锁成功，但是响应失败，需要解锁所有节点
            unlockInner(Arrays.asList(lock));
            // 表示获取锁失败
            lockAcquired = false;
        } catch (Exception e) {
            // 表示获取锁失败
            lockAcquired = false;
        }
        if (lockAcquired) {
            // 如果当前redis节点加锁成功，则加入到acquiredLocks集合中
            acquiredLocks.add(lock);
        } else {
            // 计算已经申请锁失败的节点是否已经到达 允许加锁失败节点个数限制 （N-(N/2+1)）, 如果已经到达，就认定最终申请锁失败，则没有必要继续从后面的节点申请了。因为 Redlock 算法要求至少N/2+1 个节点都加锁成功，才算最终的锁申请成功
            if (locks.size() - acquiredLocks.size() == failedLocksLimit()) {
                break;
            }
            if (failedLocksLimit == 0) {
                unlockInner(acquiredLocks);
                if (waitTime <= 0) {
                    return false;
                }
                failedLocksLimit = failedLocksLimit();
                acquiredLocks.clear();
                // reset iterator
                while (iterator.hasPrevious()) {
                    iterator.previous();
                }
            } else {
                failedLocksLimit--;
            }
        }
        // 计算 目前从各个节点获取锁已经消耗的总时间，如果已经等于最大等待时间，则认定最终申请锁失败，返回false
        if (remainTime > 0) {
            // remainTime: 锁剩余时间，这个时间是某个客户端向所有redis节点申请获取锁的总等待时间, 获取锁的中耗时时间不能大于这个时间。
            // System.currentTimeMillis() - time: 这个计算出来的就是当前redis节点获取锁消耗的时间
            remainTime -= System.currentTimeMillis() - time;
            // 重置time为当前时间，因为下一次循环的时候，方便计算下一个redis节点获取锁消耗的时间
            time = System.currentTimeMillis();
            // 锁剩余时间减到0了，说明达到最大等待时间，加锁超时，认为获取锁失败，需要对成功加锁集合 acquiredLocks 中的所有锁执行锁释放
            if (remainTime <= 0) {
                unlockInner(acquiredLocks);
                // 直接返回false，获取锁失败
                return false;
            }
        }
    }
    if (leaseTime > 0) {
        // 重置锁过期时间
        acquiredLocks.stream()
                .map(l -> (RedissonBaseLock) l)
                .map(l -> l.expireAsync(unit.toMillis(leaseTime), TimeUnit.MILLISECONDS))
                .forEach(f -> f.toCompletableFuture().join());
    }
    // 如果逻辑正常执行完则认为最终申请锁成功，返回true
    return true;
}

从源码中可以看到，红锁的加锁，其实就是循环所有加锁的节点，挨个执行LUA脚本加锁，对于加锁成功的那些节点，会加入到acquiredLocks集合中保存起来；如果加锁失败的话，则会判断已经申请锁失败的节点是否已经到达允许加锁失败节点个数限制（N-(N/2+1)）, 如果已经到达，就认定最终申请锁失败，则没有必要继续从后面的节点申请了。

并且，每个节点执行完tryLock()尝试获取锁之后，无论是否获取锁成功，都会判断目前从各个节点获取锁已经消耗的总时间，如果已经等于最大等待时间，则认定最终申请锁失败，需要对成功加锁集合 acquiredLocks 中的所有锁执行锁释放，然后返回false。

多机案例

理论参考来源

redis之父提出了Redlock算法解决这个问题

官网

具体

小总结

这个锁的算法实现了多redis实例的情况，相对于单redis节点来说，优点在于防止了单节点故障造成整个服务停止运行的情况且在多节点中锁的设计，及多节点同时崩溃等各种意外情况有自己独特的设计方法。

Redisson 分布式锁支持 MultiLock 机制可以将多个锁合并为一个大锁，对一个大锁进行统一的申请加锁以及释放锁。

最低保证分布式锁的有效性及安全性的要求如下：

1.互斥；任何时刻只能有一个client获取锁

2.释放死锁；即使锁定资源的服务崩溃或者分区，仍然能释放锁

3.容错性；只要多数redis节点（一半以上）在使用，client就可以获取和释放锁

网上讲的基于故障转移实现的redis主从无法真正实现Redlock:

因为redis在进行主从复制时是异步完成的，比如在clientA获取锁后，主redis复制数据到从redis过程中崩溃了，导致没有复制到从redis中，然后从redis选举出一个升级为主redis,造成新的主redis没有clientA 设置的锁，这是clientB尝试获取锁，并且能够成功获取锁，导致互斥失效；

代码参考来源

官网来源

2022年第8章第4小节

2023年第8章第4小节

2023年第8章第3小节

MultiLock多重锁

案例

docker走起3台redis的master机器，本次设置3台master各自独立无从属关系

docker run -p 6381:6379 --name redis-master-1 -d redis

docker run -p 6382:6379 --name redis-master-2 -d redis

docker run -p 6383:6379 --name redis-master-3 -d redis

执行成功见下：

进入上一步刚启动的redis容器实例

docker exec -it redis-master-1 /bin/bash 或者 docker exec -it redis-master-1 redis-cli

docker exec -it redis-master-2 /bin/bash 或者 docker exec -it redis-master-2 redis-cli

docker exec -it redis-master-3 /bin/bash 或者 docker exec -it redis-master-3 redis-cli

建Module

redis_redlock

改POM

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">

    <modelVersion>4.0.0</modelVersion>

    <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>2.3.10.RELEASE</version>
        <relativePath/> <!-- lookup parent from repository -->
    </parent>

    <groupId>com.test.redis.redlock</groupId>
    <artifactId>redis_redlock</artifactId>
    <version>0.0.1-SNAPSHOT</version>

    <properties>
        <java.version>1.8</java.version>
    </properties>

    <dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>

        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
        </dependency>

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

        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
            <version>1.18.8</version>
        </dependency>

        <!--swagger-->
        <dependency>
            <groupId>io.springfox</groupId>
            <artifactId>springfox-swagger2</artifactId>
            <version>2.9.2</version>
        </dependency>

        <!--swagger-ui-->
        <dependency>
            <groupId>io.springfox</groupId>
            <artifactId>springfox-swagger-ui</artifactId>
            <version>2.9.2</version>
        </dependency>

        <dependency>
            <groupId>org.apache.commons</groupId>
            <artifactId>commons-lang3</artifactId>
            <version>3.4</version>
            <scope>compile</scope>
        </dependency>

        <dependency>
            <groupId>cn.hutool</groupId>
            <artifactId>hutool-all</artifactId>
            <version>5.8.11</version>
        </dependency>

        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-configuration-processor</artifactId>
            <optional>true</optional>
        </dependency>
    </dependencies>

    <build>
        <plugins>
            <plugin>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-maven-plugin</artifactId>
                <configuration>
                    <excludes>
                        <exclude>
                            <groupId>org.springframework.boot</groupId>
                            <artifactId>spring-boot-configuration-processor</artifactId>
                        </exclude>
                    </excludes>
                </configuration>
            </plugin>
        </plugins>
    </build>

</project>

写YML

server.port=9090
spring.application.name=redlock


spring.swagger2.enabled=true


spring.redis.database=0
spring.redis.password=
spring.redis.timeout=3000
spring.redis.mode=single

spring.redis.pool.conn-timeout=3000
spring.redis.pool.so-timeout=3000
spring.redis.pool.size=10

spring.redis.single.address1=192.168.111.185:6381
spring.redis.single.address2=192.168.111.185:6382
spring.redis.single.address3=192.168.111.185:6383

主启动

package com.test.redis.redlock;

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;

@SpringBootApplication
public class RedisRedlockApplication {

    public static void main(String[] args) {
        SpringApplication.run(RedisRedlockApplication.class, args);
    }

}

业务类

配置

CacheConfiguration

package com.test.redis.redlock.config;

import org.apache.commons.lang3.StringUtils;
import org.redisson.Redisson;
import org.redisson.api.RedissonClient;
import org.redisson.config.*;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;

@Configuration
@EnableConfigurationProperties(RedisProperties.class)
public class CacheConfiguration {

    @Autowired
    private RedisProperties redisProperties;

    @Bean
    RedissonClient redissonClient1() {
        Config config = new Config();
        String node = redisProperties.getSingle().getAddress1();
        node = node.startsWith("redis://") ? node : "redis://" + node;
        SingleServerConfig serverConfig = config.useSingleServer()
                .setAddress(node)
                .setTimeout(redisProperties.getPool().getConnTimeout())
                .setConnectionPoolSize(redisProperties.getPool().getSize())
                .setConnectionMinimumIdleSize(redisProperties.getPool().getMinIdle());

        if (StringUtils.isNotBlank(redisProperties.getPassword())) {
            serverConfig.setPassword(redisProperties.getPassword());
        }

        return Redisson.create(config);
    }

    @Bean
    RedissonClient redissonClient2() {
        Config config = new Config();
        String node = redisProperties.getSingle().getAddress2();
        node = node.startsWith("redis://") ? node : "redis://" + node;
        SingleServerConfig serverConfig = config.useSingleServer()
                .setAddress(node)
                .setTimeout(redisProperties.getPool().getConnTimeout())
                .setConnectionPoolSize(redisProperties.getPool().getSize())
                .setConnectionMinimumIdleSize(redisProperties.getPool().getMinIdle());

        if (StringUtils.isNotBlank(redisProperties.getPassword())) {
            serverConfig.setPassword(redisProperties.getPassword());
        }

        return Redisson.create(config);
    }

    @Bean
    RedissonClient redissonClient3() {
        Config config = new Config();
        String node = redisProperties.getSingle().getAddress3();
        node = node.startsWith("redis://") ? node : "redis://" + node;
        SingleServerConfig serverConfig = config.useSingleServer()
                .setAddress(node)
                .setTimeout(redisProperties.getPool().getConnTimeout())
                .setConnectionPoolSize(redisProperties.getPool().getSize())
                .setConnectionMinimumIdleSize(redisProperties.getPool().getMinIdle());

        if (StringUtils.isNotBlank(redisProperties.getPassword())) {
            serverConfig.setPassword(redisProperties.getPassword());
        }

        return Redisson.create(config);
    }


    /**
     * 单机
     * @return redisson对象
     */
    /* @Bean
     * public Redisson redisson() {
     *   Config config = new Config();          
     *   config.useSingleServer().setAddress("redis://192.168.111.147:6379").setDatabase(0);
     *   return (Redisson) Redisson.create(config);
    }*/

}

RedisPoolProperties

package com.test.redis.redlock.config;

import lombok.Data;

@Data
public class RedisPoolProperties {

    private int maxIdle;

    private int minIdle;

    private int maxActive;

    private int maxWait;

    private int connTimeout;

    private int soTimeout;

    /**
     * 池大小
     */
    private  int size;

}

RedisProperties

package com.test.redis.redlock.config;

import lombok.Data;
import lombok.ToString;
import org.springframework.boot.context.properties.ConfigurationProperties;

@Data
@ConfigurationProperties(prefix = "spring.redis", ignoreUnknownFields = false)
public class RedisProperties {

    private int database;

    /**
     * 等待节点回复命令的时间。该时间从命令发送成功时开始计时
     */
    private int timeout;

    private String password;

    private String mode;

    /**
     * 池配置
     */
    private RedisPoolProperties pool;

    /**
     * 单机信息配置
     */
    private RedisSingleProperties single;

}

RedisSingleProperties

package com.test.redis.redlock.config;

import lombok.Data;

@Data
public class RedisSingleProperties {

    private  String address1;

    private  String address2;

    private  String address3;

}

controller

package com.test.redis.redlock.controller;

import cn.hutool.core.util.IdUtil;
import lombok.extern.slf4j.Slf4j;
import org.redisson.Redisson;
import org.redisson.RedissonMultiLock;
import org.redisson.RedissonRedLock;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.redisson.config.Config;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.data.redis.RedisProperties;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;

import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

@Slf4j
@RestController
public class RedLockController {

    public static final String CACHE_KEY_REDLOCK = "ATGUIGU_REDLOCK";

    @Autowired
    RedissonClient redissonClient1;

    @Autowired
    RedissonClient redissonClient2;

    @Autowired
    RedissonClient redissonClient3;

    boolean isLockBoolean;

    @GetMapping(value = "/multiLock")
    public String getMultiLock() throws InterruptedException {
        String uuid =  IdUtil.simpleUUID();
        String uuidValue = uuid+":" + Thread.currentThread().getId();

        RLock lock1 = redissonClient1.getLock(CACHE_KEY_REDLOCK);
        RLock lock2 = redissonClient2.getLock(CACHE_KEY_REDLOCK);
        RLock lock3 = redissonClient3.getLock(CACHE_KEY_REDLOCK);

        RedissonMultiLock redLock = new RedissonMultiLock(lock1, lock2, lock3);
        redLock.lock();

        try {
            System.out.println(uuidValue + "\t" + "---come in biz multiLock");

            try { 
                TimeUnit.SECONDS.sleep(30); 
            } catch (InterruptedException e) {
                e.printStackTrace(); 
            }

            System.out.println(uuidValue + "\t" + "---task is over multiLock");
        } catch (Exception e) {
            e.printStackTrace();
            log.error("multiLock exception ", e);
        } finally {
            redLock.unlock();
            log.info("释放分布式锁成功key:{}", CACHE_KEY_REDLOCK);
        }

        return "multiLock task is over  " + uuidValue;
    }

}

测试

http://localhost:9090/multilock

命令

TTL ATGUIGU_REDLOCK

HGETALL ATGUIGU_REDLOCK

shutdown

docker start redis-master-1

docker exec -it redis-master-1 redis-cli

结论

RedLock算法问题

1、持久化问题

假设一共有5个Redis节点：A, B, C, D, E：

客户端1成功锁住了A, B, C，获取锁成功，但D和E没有锁住。

节点C崩溃重启了，但客户端1在C上加的锁没有持久化下来，丢失了。

节点C重启后，客户端2锁住了C, D, E，获取锁成功。

这样，客户端1和客户端2同时获得了锁（针对同一资源）。

2、客户端长时间阻塞，导致获得的锁释放，访问的共享资源不受保护的问题。

3、Redlock算法对时钟依赖性太强，若某个节点中发生时间跳跃（系统时间戳不正确），也可能会引此而引发锁安全性问题。

总结

红锁其实也并不能解决根本问题，只是降低问题发生的概率。完全相互独立的redis，每一台至少也要保证高可用，还是会有主从节点。既然有主从节点，在持续的高并发下，master还是可能会宕机，从节点可能还没来得及同步锁的数据。很有可能多个主节点也发生这样的情况，那么问题还是回到一开始的问题，红锁只是降低了发生的概率。

其实，在实际场景中，红锁是很少使用的。这是因为使用了红锁后会影响高并发环境下的性能，使得程序的体验更差。所以，在实际场景中，我们一般都是要保证Redis集群的可靠性。同时，使用红锁后，当加锁成功的RLock个数不超过总数的一半时，会返回加锁失败，即使在业务层面任务加锁成功了，但是红锁也会返回加锁失败的结果。另外，使用红锁时，需要提供多套Redis的主从部署架构，同时，这多套Redis主从架构中的Master节点必须都是独立的，相互之间没有任何数据交互。

Redis的缓存过期淘汰策略

Redis内存满了怎么办

redis默认内存多少？在哪里查看?如何设置修改?

查看Redis最大占用内存

打开redis配置文件，设置maxmemory参数，maxmemory是bytes字节类型，注意转换。

redis默认内存多少可以用？

注意，在 64bit 系统下，maxmemory 设置为 0 表示不限制 Redis 内存使用

一般生产上你如何配置？

一般推荐Redis设置内存为最大物理内存的四分之三。

如何修改redis内存设置

通过修改文件配置

通过命令修改

什么命令查看redis内存使用情况?

info memory

config get maxmemory

真要打满了会怎么样？如果Redis内存使用超出了设置的最大值会怎样？

改改配置，故意把最大值设为1个byte试试

结论

设置了maxmemory的选项，假如redis内存使用达到上限

没有加上过期时间就会导致数据写满maxmemory为了避免类似情况，引出下一章内存淘汰策略。

往redis里写的数据是怎么没了的?它如何删除的？

redis过期键的删除策略

如果一个键是过期的，那它到了过期时间之后是不是马上就从内存中被被删除呢？？

如果回答yes，立即删除，那是错误的？

如果不是，那过期后到底什么时候被删除呢？？是个什么操作？

三种不同的删除策略

立即删除

Redis不可能时时刻刻遍历所有被设置了生存时间的key，来检测数据是否已经到达过期时间，然后对它进行删除。

立即删除能保证内存中数据的最大新鲜度，因为它保证过期键值会在过期后马上被删除，其所占用的内存也会随之释放。但是立即删除对cpu是最不友好的。因为删除操作会占用cpu的时间，如果刚好碰上了cpu很忙的时候，比如正在做交集或排序等计算的时候，就会给cpu造成额外的压力，让CPU心累，时时需要删除，忙死。。。。。。。

这会产生大量的性能消耗，同时也会影响数据的读取操作。

总结：对CPU不友好，用处理器性能换取存储空间（拿时间换空间）

惰性删除

数据到达过期时间，不做处理。等下次访问该数据时，

如果未过期，返回数据；

发现已过期，删除，返回不存在。

惰性删除策略的缺点是，它对内存是最不友好的。

如果一个键已经过期，而这个键又仍然保留在redis中，那么只要这个过期键不被删除，它所占用的内存就不会释放。

在使用惰性删除策略时，如果数据库中有非常多的过期键，而这些过期键又恰好没有被访问到的话，那么它们也许永远也不会被删除(除非用户手动执行FLUSHDB)，我们甚至可以将这种情况看作是一种内存泄漏–无用的垃圾数据占用了大量的内存，而服务器却不会自己去释放它们，这对于运行状态非常依赖于内存的Redis服务器来说,肯定不是一个好消息。

总结：对memory不友好，用存储空间换取处理器性能（拿空间换时间）

开启憜性淘汰，lazyfree-lazy-eviction=yes

上面两种方案都走极端

定期删除

定期删除策略是前两种策略的折中：

定期删除策略每隔一段时间执行一次删除过期键操作并通过限制删除操作执行时长和频率来减少删除操作对CPU时间的影响。

周期性轮询redis库中的时效性数据，采用随机抽取的策略，利用过期数据占比的方式控制删除频度

特点1：CPU性能占用设置有峰值，检测频度可自定义设置

特点2：内存压力不是很大，长期占用内存的冷数据会被持续清理

总结：周期性抽查存储空间（随机抽查，重点抽查）

举例：

redis默认每隔100ms检查是否有过期的key，有过期key则删除。注意：redis不是每隔100ms将所有的key检查一次而是随机抽取进行检查( 如果每隔100ms,全部key进行检查，redis直接进去ICU)。因此，如果只采用定期删除策略，会导致很多key到时间没有删除。

定期删除策略的难点是确定删除操作执行的时长和频率：如果删除操作执行得太频繁或者执行的时间太长，定期删除策略就会退化成立即删除策略，以至于将CPU时间过多地消耗在删除过期键上面。如果删除操作执行得太少，或者执行的时间太短，定期删除策略又会和惰性删除束略一样，出现浪费内存的情况。因此，如果采用定期删除策略的话，服务器必须根据情况，合理地设置删除操作的执行时长和执行频率。

定期抽样key，判断是否过期

但是会存在漏网之鱼

上述步骤都过堂了，还有漏洞吗?

1 定期删除时，从来没有被抽查到

2 惰性删除时，也从来没有被点中使用过

上述两个步骤======> 大量过期的key堆积在内存中，导致redis内存空间紧张或者很快耗尽

必须要有一个更好的兜底方案。。。。。。

redis缓存淘汰策略登场。。。。。。O(∩_∩)O哈哈~

redis缓存淘汰策略

redis配置文件

【MEMORY MANAGEMENT】