【1】. 使用场景:多个服务间保证同一时刻同一时间段内同一用户只能有一个请求(防止关键业务出现并发攻击)
【2】. Redis分布式锁比较正确的姿势是采用redisson这个客户端工具

①. 如何基于官网进行开发

• ①. 进入redis中文官网,点击文档
  
• ②.选择分布式锁,打开页面
  
• ③. 点击Wiki
  
  
• ④. 后续关于分布式锁,需要什么内容,进行文档的查阅
  
• ⑤. 天上飞的理念(RedLock)必然有落地的实现(Redisson)
  
• ⑥. redission解决了两个问题

1. 锁的自动续期,如果业务超长,运行期间自动给锁续上新的30s,不用担心业务时间长
2. 如果业务宕机了,这个默认的过期时间是30s,避免了死锁
3. 加锁的业务只要运行完成,就不会给当前锁续期,即使不手动解锁,锁默认在30s以后自动删除

②. 单机案例(了解即可)

三个重要元素:加锁、解锁、超时

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

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

// 为了保证解锁操作的原子性,我们用LUA脚本完成这一操作。先判断当前锁的字符串是否与传入的值相等,是的话就删除Key,解锁成功。
if redis.call('get',KEYS[1]) == ARGV[1] then 
   return redis.call('del',KEYS[1]) 
else
   return 0 
end

• ③. 超时:锁key要注意过期时间,不能长期占用

• ④. 加锁和解锁的关键代码逻辑

// 加锁
public static boolean tryLock(String key, String uniqueId, int seconds) {
    return "OK".equals(jedis.set(key, uniqueId, "NX", "EX", seconds));
}
// 解锁
public static boolean releaseLock(String key, String uniqueId) {
    String luaScript = "if redis.call('get', KEYS[1]) == ARGV[1] then " +
            "return redis.call('del', KEYS[1]) else return 0 end";
    
    return jedis.eval(
            luaScript,
            Collections.singletonList(key),
            Collections.singletonList(uniqueId)
    ).equals(1L);
}

③. setnx的分布式锁有哪些不足

• ①. 基于setnx的分布式锁有什么缺点？

1. 线程1首先获取锁成功,将键值对写入redis的master节点
2. 在redis将该键值对同步到slave节点之前,master发生了故障
3. redis触发故障转移,其中一个slave升级为新的master
4. 此时新的master并不包含线程1写入的键值对,因此线程2尝试获取锁也可以成功拿到锁
5. 此时相当于有两个线程获取到了锁,可能会导致各种预期之外的情况发生,例如最常见的脏数据。
   我们加的是排它独占锁,同一时间只能有一个建redis锁成功并持有锁,严禁出现2个以上的请求线程拿到锁。
   

• ②. redis之父提出了Redlock算法解决这个问题
  (Redis也提供了Redlock算法,用来实现基于多个实例的分布式锁。锁变量由多个实例维护,即使有实例发生了故障,锁变量仍然是存在的,客户端还是可以完成锁操作。Redlock算法是实现高可靠分布式锁的一种有效解决方案,可以在实际开发中使用)
  

• ③. Redis集群的AP(redis异步复制造成的锁丢失,比如:主节点没来的及把刚刚set进来这条数据给从节点,就挂了)

• ④. redis分布式锁,多主集群模式,需要计算容错率(N=2X+1)

1. 比如,网络中死了1台机器,我要求还是OK的,可以用,请问,最多主集群部署几台?
   N=2*1+1=3
2. N表示部署台数、X表示宕机数
   

• ⑤. 为什么是奇数？ N = 2X + 1 (N是最终部署机器数,X是容错机器数)
  最少的机器,最多的产出效果
  加入在集群环境中,redis失败1台,可接受。2N+2= 2 * 1+2 =4,部署4台
  加入在集群环境中,redis失败2台,可接受。2N+2 = 2 * 2+2 =6,部署6台

• ⑥. 那么什么是容错呢?

1. 失败了多少个机器实例后我还是可以容忍的,所谓的容忍就是数据一致性还是可以Ok的,CP数据一致性还是可以满足
2. 加入在集群环境中,redis失败1台,可接受。2X+1 = 2 * 1+1 =3,部署3台,死了1个剩下2个可以正常工作,那就部署3台
3. 加入在集群环境中,redis失败2台,可接受。2X+1 = 2 * 2+1 =5,部署5台,死了2个剩下3个可以正常工作,那就部署5台

④. 基于Redisson的入门案例

• ①. 导入pom

	<!--引入redisson分布式锁-->
	<dependency>
		<groupId>org.redisson</groupId>
		<artifactId>redisson</artifactId>
		<version>3.13.4</version>
	</dependency>

• ②. 建立配置类,参考文档

//2.建立配置类
@Configuration
public class MyRedisConfig {
    /**
     * 所有对Redisson的使用
     * @return
     * @throws IOException
     */
    //https://github.com/redisson/redisson/wiki/14.-%E7%AC%AC%E4%B8%89%E6%96%B9%E6%A1%86%E6%9E%B6%E6%95%B4%E5%90%88
    @Bean(destroyMethod="shutdown")
    public RedissonClient redisson() throws IOException {
        Config config = new Config();
        // 创建单例模式的配置
        //config.useSingleServer().setAddress("redis://" + ipAddr + ":6379");
        config.useSingleServer().setAddress("redis://192.168.56.10:6379");
        return Redisson.create(config);
    }
}

• ③. 进行单元测试

@Autowired
//RedissonClient redissonClient;
RedissonClient redisson;
@Test
public void redission(){
	System.out.println(redissonClient);
	RLock lock = redisson.getLock("lock");
}

⑤. 三台主机案例搭建

• ①. docker上安装三台机器

 docker run -p 6381:6379 --name redis-master-1 -d redis:6.08
docker run -p 6382:6379 --name redis-master-2 -d redis:6.0.8
docker run -p 6383:6379 --name redis-master-3 -d redis:6.0.8

• ②. 建Module(redis_redlock)、改pom.xml

<?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.xiaozhi.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.12.0</version>-->
            <version>3.13.4</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>

    </dependencies>

    <build>
        <plugins>
            <plugin>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-maven-plugin</artifactId>
            </plugin>
        </plugins>
    </build>
</project>

• ③. 各个配置文件如下

spring.application.name=spring-boot-redis
server.port=9090

spring.swagger2.enabled=true

spring.redis.database=0
spring.redis.password=
spring.redis.timeout=3000
#sentinel/cluster/single
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.147:6381
spring.redis.single.address2=192.168.111.147:6382
spring.redis.single.address3=192.168.111.147:6383

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

    @Autowired
    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
     */
    /*@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
@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
@ConfigurationProperties(prefix = "spring.redis", ignoreUnknownFields = false)
@Data
public class RedisProperties {
    private int database;
    /**
     * 等待节点回复命令的时间。该时间从命令发送成功时开始计时
     */
    private int timeout;
    private String password
    private String mode;
    /**
     * 池配置
     */
    private RedisPoolProperties pool;
    /**
     * 单机信息配置
     */
    private RedisSingleProperties single;
}

//RedisSingleProperties
@Data
public class RedisSingleProperties {
    private  String address1;
    private  String address2;
    private  String address3;
}

• ④. controller代码展示

@RestController
@Slf4j
public class RedLockController {
    public static final String CACHE_KEY_REDLOCK = "TANGZHI_REDLOCK";
    @Autowired
    RedissonClient redissonClient1;
    @Autowired
    RedissonClient redissonClient2;
    @Autowired
    RedissonClient redissonClient3;
    @GetMapping(value = "/redlock")
    public void getlock() {
        //CACHE_KEY_REDLOCK为redis 分布式锁的key
        RLock lock1 = redissonClient1.getLock(CACHE_KEY_REDLOCK);
        RLock lock2 = redissonClient2.getLock(CACHE_KEY_REDLOCK);
        RLock lock3 = redissonClient3.getLock(CACHE_KEY_REDLOCK);
        RedissonRedLock redLock = new RedissonRedLock(lock1, lock2, lock3);
        boolean isLockBoolean;
        try {
            //waitTime 抢锁的等待时间,正常情况下等3秒
            //leaseTime就是redis key的过期时间,正常情况下等5分钟300秒。
            isLockBoolean = redLock.tryLock(3, 300, TimeUnit.SECONDS);
            log.info("线程{},是否拿到锁:{} ",Thread.currentThread().getName(),isLockBoolean);
            if (isLockBoolean) {
                System.out.println(Thread.currentThread().getName()+"\t"+"---come in biz");
                //业务逻辑,忙10分钟
                try { TimeUnit.MINUTES.sleep(10); } catch (InterruptedException e) { e.printStackTrace(); }
            }
        } catch (Exception e) {
            log.error("redlock exception ",e);
        } finally {
            // 无论如何, 最后都要解锁
            redLock.unlock();
        }
    }
}

• ⑤. 测试:http://localhost:9090/redlock

[root@TANG2021 ~]# docker exec -it redis-master-1 redis-cli
127.0.0.1:6379> keys *
1) "TANGZHI_REDLOCK"
127.0.0.1:6379> type TANGZHI_REDLOCK
hash
127.0.0.1:6379> hgetall TANGZHI_REDLOCK
1) "ca512c4b-f05f-4578-9b8f-45e2e35aa0d7:50"
2) "1"

⑥. Redisson源码解析

• ①. 测试代码展示

public class WatchDogDemo {
    public static final String LOCKKEY = "DEBUG_YUANMA";

    private static Config config;
    private static Redisson redisson;

    static {
        config = new Config();
        config.useSingleServer().setAddress("redis://"+"192.168.68.143"+":6379").setDatabase(0);
        redisson = (Redisson)Redisson.create(config);
    }

    public static void main(String[] args) {
        RLock redissonLock = redisson.getLock(LOCKKEY);
        redissonLock.lock();
        try {
            System.out.println("1111-------biz");
            //暂停几秒钟线程
            try { TimeUnit.SECONDS.sleep(25); } catch (InterruptedException e) { e.printStackTrace(); }
        }catch (Exception e){
            e.printStackTrace();
        }finally {
            if(redissonLock.isLocked() && redissonLock.isHeldByCurrentThread()) {
                redissonLock.unlock();
            }
        }
        System.out.println(Thread.currentThread().getName() + " main ------ ends.");
        //暂停几秒钟线程
        try { TimeUnit.SECONDS.sleep(3); } catch (InterruptedException e) { e.printStackTrace(); }
        redisson.shutdown();
    }
}

• ②. Redis分布式锁过期了,但是业务逻辑还没处理完怎么办?引入缓存续命

1. 额外起一个线程,定期检查线程是否还持有锁,如果有则延长过期时间。
2. Redisson里面就实现了这个方案,使用“看门狗”定期检查(每1/3的锁时间检查1次),如果线程还持有锁,则刷新过期时间
3. 在获取锁成功后,给锁加一个watchdog,watchdog 会起一个定时任务,在锁没有被释放且快要过期的时候会续期
   

• ③. 详解缓存续命源码解析一:通过redisson新建出来的锁key,默认是30秒
  
• ④. 详解缓存续命源码解析二
  
• ⑤. 详解缓存续命源码解析三

1. 这里面初始化了一个定时器,dely的时间是 internalLockLeaseTime/3
2. 在Redisson中,internalLockLeaseTime是30s,也就是每隔10s续期一次,每次30s

• ⑥. watch dog自动延期机制
  (客户端A加锁成功,就会启动一个watch dog看门狗,它是一个后台线程,会每隔10秒检查一下,如果客户端A还持有锁key,那么就会不断的延长锁key的生存时间,默认每次续命又从30秒新开始)
  
• ⑦. 详解缓存续命源码解析四:加锁逻辑(针对于默认时间30s过期的) lock.lock()

1. 没有锁,加锁(唯一的id),启动定时任务,设置30s的过期时间
2. 有锁:将唯一id+1(可重入锁)
3. 非加锁线程,返回当前加锁线程的过期时间
   

• ⑧. 详解缓存续命源码解析五:解锁逻辑

1. 如果释放的锁的线程和已存在锁的线程不是同一个线程,返回null
2. 通过hincrby递减1,先释放一次锁。若剩余次数还大于0,则证明当前锁是可重入锁,刷新过期时间
3. 若剩余次数小于0,删除key并释放锁,解锁成功

• ⑨. 详解缓存续命源码解析六:加锁逻辑
  (lock.lock(10,TimeUnit.SECONDS):10s自动解锁,不会自动续期)
  底层原理:如果我们传递了锁的超时时间,就发送给redis执行lua脚本,进行占锁,默认超时就是我们指定的时间

    <T> RFuture<T> tryLockInnerAsync(long waitTime, long leaseTime, TimeUnit unit, long threadId, RedisStrictCommand<T> command) {
        this.internalLockLeaseTime = unit.toMillis(leaseTime);
        return this.evalWriteAsync(this.getName(), LongCodec.INSTANCE,
 command, "if (redis.call('exists', KEYS[1]) == 0) then redis.call('hincrby', KEYS[1], ARGV[2], 1); redis.call('pexpire', KEYS[1], ARGV[1]); return nil; end; if (redis.call('hexists', KEYS[1], ARGV[2]) == 1) then redis.call('hincrby', KEYS[1], ARGV[2], 1); redis.call('pexpire', KEYS[1], ARGV[1]); return nil; end; return redis.call('pttl', KEYS[1]);", Collections.singletonList(this.getName()), this.internalLockLeaseTime, this.getLockName(threadId));
    }

• ⑩. 可重入锁
  Key为锁的名称,field为随机字符串+线程ID,值为1

⑦. 常见的异常情况解析

• ①. 使用分布式锁的时候,业务执行时长为20s,自己设置了分布式锁的时间是10s,这个时候先进行lock上锁操作,最后程序结束后,进行unlock就会报错
  
  
  

• ②. 针对上面的解决方案如下:
  

• ③. 最终代码

    @GetMapping("/buy_goods")
    public String buy_Goods() throws IOException{
        RLock redissonLock = redisson.getLock(REDIS_LOCK_KEY);
        redissonLock.lock();
        try{
            String result = stringRedisTemplate.opsForValue().get("goods:001");
            int goodsNumber = result == null ? 0 : Integer.parseInt(result);

            if(goodsNumber > 0){
                int realNumber = goodsNumber - 1;
                stringRedisTemplate.opsForValue().set("goods:001",realNumber + "");
                System.out.println("你已经成功秒杀商品,此时还剩余:" + realNumber + "件"+"\t 服务器端口:"+serverPort);
                //暂停几秒钟线程
                try { TimeUnit.SECONDS.sleep(5); } catch (InterruptedException e) { e.printStackTrace(); }
                return "你已经成功秒杀商品,此时还剩余:" + realNumber + "件"+"\t 服务器端口:"+serverPort;
            }else{
                System.out.println("商品已经售罄/活动结束/调用超时,欢迎下次光临"+"\t 服务器端口:"+serverPort);
            }
            return "商品已经售罄/活动结束/调用超时,欢迎下次光临"+"\t 服务器端口:"+serverPort;
        }finally {
            if(redissonLock.isLocked() && redissonLock.isHeldByCurrentThread()){
                redissonLock.unlock();
            }
        }
    }

⑧. 分布式锁应用场景

一句话记忆：
“高并发 + 共享资源 + 只能一人动”——就上分布式锁

• ①. 用户领红包场景
  领红包场景下面,防止红包多抢情况发生

• ②. 三方同步订单,创建订单后,出现拆单的情况下,只同步一笔

• ③. 多节点部署的情况下,调度时间周期频繁,为了不对数据产生影响

• ④. 接口防重,多节点情况下,只能有一个执行成功

• ⑤. 消息去重、幂等处理
  Kafka/RocketMQ 并发消费时,同一条消息只被处理一次

• ⑥. 并发减库存(超卖克星)
  秒杀/拼团，多机器同时扣同一商品库存,必须串行扣