分布式锁

Author：Exchanges
Version：9.0.0

一、引言

高并发的情况下还要保证数据的安全性问题：在互联网一些秒杀的环境下，例如：抢优惠券啊，秒杀商品等等，如果处理不当会产生超卖现象，因为是分布式环境，传统的一些技术会失败，比如传统的synchronized或者lock锁，以及数据库的事务，会无法保证ACID，我们需要想办法去解决，这里我们使用Redisson和Seata来解决分布式架构中锁和事务的问题

二、分布式锁介绍

由于传统的锁是基于Tomcat服务器内部的，搭建了集群之后，导致锁失效，使用分布式锁来处理。

分布式锁介绍

三、分布式锁解决方案

3.1 搭建环境

1.创建SpringBoot工程并导入依赖

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-data-redis-reactive</artifactId>
    <version>2.3.7.RELEASE</version>
</dependency>

2.配置application.properties文件，然后启动redis-cli，执行：set stock 1000

#配置redis
spring.redis.host=127.0.0.1
spring.redis.database=0
spring.redis.port=6379

3.编写抢购业务的SecondKillController

package com.qf.controller;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

@RestController
@RequestMapping("secondkill")
public class SecondKillController {

    @Autowired
    private StringRedisTemplate stringRedisTemplate;

    @RequestMapping("stock")
    public String stock(){
        //获取库存数量
        Integer stock = Integer.parseInt(stringRedisTemplate.opsForValue().get("stock"));
        //判断
        if(stock>0){
            int realStock = stock - 1;
            //重新设置库存数量
            stringRedisTemplate.opsForValue().set("stock",realStock+"");
            System.out.println("抢购成功！剩余库存数为："+realStock);
        }else {
            System.out.println("商品库存数不足！");
        }
        return "success";
    }
}

3.启动工程，使用 Jmter 进行压力测试，会发现数据出现不一致现象（超卖现象）
图片.png
4.我们可以使用synchronized关键字，但在tomcat集群环境下依然会有此现象，如需测试，需要配置一台Nginx以及两台Tomcat

3.2 Redis实现分布式锁原理

Redis实现分布式锁原理

3.3 Redis实现分布式锁

1.修改SecondKillController

package com.qf.controller;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.util.UUID;
import java.util.concurrent.TimeUnit;

@RestController
@RequestMapping("secondkill")
public class SecondKillController {

    @Autowired
    private StringRedisTemplate stringRedisTemplate;

    @RequestMapping("stock")
    public String stock(){

        //设置该商品的key
        String productKey = "product_001";
        //不同的线程设置不同的值
        String productVal = UUID.randomUUID().toString();

        try{
            //redis实现分布式锁，使用setIfAbsent方法（等同于redis中的setnx命令）
            Boolean product001 = stringRedisTemplate.opsForValue().
                    setIfAbsent(productKey, productVal, 10, TimeUnit.SECONDS);

            if(!product001){
                System.out.println("该商品抢购繁忙，请稍后再试");
            }

            //获取库存数量
            Integer stock = Integer.parseInt(stringRedisTemplate.opsForValue().get("stock"));
            //判断
            if(stock>0){
                int realStock = stock - 1;
                //重新设置库存数量
                stringRedisTemplate.opsForValue().set("stock",realStock+"");
                System.out.println("抢购成功！剩余库存数为："+realStock);
            }else {
                System.out.println("商品库存数不足！");
            }

        }finally {
            //删除该商品的key
            if(productVal.equals(stringRedisTemplate.opsForValue().get(productKey))){
                stringRedisTemplate.delete(productKey);
            }
        }
        return "success";
    }
}

2.重启工程，使用 Jmter 进行压力测试，并发数据量大的情况下依旧会有超卖问题

四、Redisson

4.1 介绍

Redisson是一个基于NIO的Netty框架的企业级的开源Redis Client，也提供了分布式锁的支持，Redisson 是架设在 Redis 基础上的一个 Java 驻内存数据网格框架, 充分利用 Redis 键值数据库提供的一系列优势, 基于 Java 实用工具包中常用接口, 为使用者提供了一系列具有分布式特性的常用工具类.

4.2 Redisson的使用

1.导入依赖

<dependency>
    <groupId>org.redisson</groupId>
    <artifactId>redisson-spring-boot-starter</artifactId>
    <version>3.13.6</version>
</dependency>

2.创建redisson配置类

package com.qf.config;

import org.redisson.Redisson;
import org.redisson.api.RedissonClient;
import org.redisson.config.Config;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class RedissonConfig {

    @Bean
    public RedissonClient redissonClient(){
        Config config = new Config();
        config.useSingleServer().setAddress("redis://127.0.0.1:6379").setDatabase(0);
        RedissonClient redissonClient = Redisson.create(config);

        return redissonClient;
    }
}

3.修改SecondKillController

package com.qf.controller;

import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.util.UUID;
import java.util.concurrent.TimeUnit;

@RestController
@RequestMapping("secondkill")
public class SecondKillController {

    @Autowired
    private RedissonClient redissonClient;

    @Autowired
    private StringRedisTemplate stringRedisTemplate;

    @RequestMapping("stock")
    public String stock(){

        //设置该商品的key
        String productKey = "product_001";
        //不同的线程设置不同的值
        //String productVal = UUID.randomUUID().toString();

        //使用redisson获取锁
        RLock lock = redissonClient.getLock(productKey);

        try{
            //redis实现分布式锁，使用setIfAbsent方法（等同于redis中的setnx命令）
//            Boolean product001 = stringRedisTemplate.opsForValue().
//                    setIfAbsent(productKey, productVal, 10, TimeUnit.SECONDS);
//
//            if(!product001){
//                System.out.println("该商品抢购繁忙，请稍后再试");
//            }

            //加锁
            lock.lock();

            //获取库存数量
            Integer stock = Integer.parseInt(stringRedisTemplate.opsForValue().get("stock"));
            //判断
            if(stock>0){
                int realStock = stock - 1;
                //重新设置库存数量
                stringRedisTemplate.opsForValue().set("stock",realStock+"");
                System.out.println("抢购成功！剩余库存数为："+realStock);
            }else {
                System.out.println("商品库存数不足！");
            }

        }finally {

            //释放锁
            lock.unlock();

            //删除该商品的key
//            if(productVal.equals(stringRedisTemplate.opsForValue().get(productKey))){
//                stringRedisTemplate.delete(productKey);
//            }
        }
        return "success";
    }
}

3.重启工程，使用 Jmter 再次进行压力测试

4.3Redisson源码分析

源码分析图
图片.png
核心源码，基于Lua脚本语言（具有原子性）

4.4 面试题：Redis和Zookeeper实现分布式锁的区别？

在CAP定理中： Redis： AP（保证可用性和分区容错性），Master加锁成功后就给客户端返回成功标识了，而不是等到同步完Slave再发，如果Slave还没同步完成，此时Maste宕机了，就会有问题。 Zookeeper ： CP （保证一致性和分区容错性），如果要保证完全一致性，可以使用zookeeper锁，CP，能够保证绝对一致， Zookeeper是Master节点加锁后，把状态也同步到了另外的节点成功后才给客户端返回成功标识但是性能没有redis高。

一、redis和zookeeper技术有何不同？（区别）

Redis 是nosql数据，主要特点缓存。
Zookeeper是分布式协调工具，主要用于分布式解决方案。

二、Redis实现分布式锁与Zookeeper实现分布式锁的思路分别是什么？（区别）

获取锁
Zookeeper：多个客户端（jvm），会在Zookeeper上创建同一个临时节点，因为Zookeeper节点命名路径保证唯一，不允许出现重复，只要谁能够先创建成功，谁能够获取到锁。 Redis：多个客户端（jvm），会在Redis使用setnx命令创建相同的一个key，因为Redis的key保证唯一，不允许出现重复，只要谁能够先创建成功，谁能够获取到锁。释放锁
Zookeeper：使用直接关闭临时节点session会话连接，因为临时节点生命周期与session会话绑定在一块，如果session会话连接关闭的话，该临时节点也会被删除。
这时候客户端使用事件监听，如果该临时节点被删除的话，重新进入盗获取锁的步骤。

Redis：在释放锁的时候，为了确保是锁的一致性问题，在删除的redis 的key时候，需要判断同一个锁的id，才可以删除。 https://blog.csdn.net/xiaoxiaole0313/article/details/107011095/

三、redis和zookeeper如何解决死锁问题？（区别）

Zookeeper使用会话有效期方式解决死锁现象。
Redis 是对key设置有效期解决死锁现象

四、分别从性能和可靠性两个角度谈谈redis和zookeeper实现分布式锁的优缺点（区别）

性能角度考虑：
因为Redis是NoSQL数据库，相对比来说Redis比Zookeeper性能要好。可靠性：
从可靠性角度分析，Zookeeper可靠性比Redis更好，因为Redis有效期不是很好控制，可能会产生有效期延迟，Zookeeper就不一样，因为Zookeeper临时节点先天性可控的有效期，所以相对来说Zookeeper比Redis更好

羊群效应

当jvm释放锁的时候，会唤醒正在等待的jvm 从新进入到获取锁的状态。
如果正在阻塞的等待获取锁的jvm，如果有几十个或者几百个、上千个的情况下
ZkServer端唤醒所有正在等待的jvm，从新进入到获取锁的状态，唤醒的成本是非常高
有可能会造成我们ZkServer端阻塞。

锁续命

业务超时,续命多次(3次)还没有释放锁,视为超时

若多次超时应该

主动释放锁；
事务回滚；
主动停止阻塞的线程；
移除key