【Redis】基于Redission实现分布式锁（代码实现）

目录

基于Redission实现分布式锁解决商品秒杀超卖的场景：

1.引入依赖：

2.加上redis的配置：

3.添加配置类：

4.编写代码实现：

5.模拟服务器分布式集群的情况：

        1.右键点击Copy Configuration

        2.点击Modify option

        3. 选择VM option(用于指定新的端口)

        4.输入想要指定的端口(比如)：-Dserver.port=8082 点击Apply

        5.出现新的进程，点击启动，就可以进行分布式多节点测试。

使用Jmeter进行压测

---

        （单机部署）多线程高并发情况下对同一个共享资源进行读写时，会出现数据错乱（数据不一致）的问题；加锁（同步锁）可以解决出现数据不一致的问题；（其他线程进行等待持有锁的线程执行完成后才能进行正常的处理）。
        但是随着用户量日益增多，单个服务器压力越来越大，所以使用多个服务器进行分布式集群部署，虽然降低了服务器的压力，提高了服务器的吞吐量。但是还是会出现数据不一致的问题，这时候发现是是同步锁（synchronized）的问题，同步锁基于JVM的，他只能锁住单个服务器中一个线程，但是经过分布式集群部署过后，每台服务器在并发的情况下只能锁住一个线程，所以高并发情况下，还是会出现数据错乱的情况。（假如4个服务器，每台服务器都处于高并发的情况，然后同步锁只能锁住一个线程，这时候每个服务器锁住一个线程，最多可以出现4个线程同时对数据库进行操作，就会造成数据不一致的问题（例如常说的秒杀超卖的情况））

        经过查阅资料发现可以通过分布式锁可以解决，然后有三种主流的分布式锁的解决方案分别是使用基于Mysql/Zookeeper/redis实现分布式锁。由于我们系统使用到了Redis，考虑Zookeeper需要重新部署到服务器，避免间接增加服务器的成本，所以直接使用Redis来实现分布式锁。我们发现使用Redis的setNX可以很简单的实现分布式锁。
        那么setNX的特性是什么呢？
        当一个线程进来，往Redis当中通过setNX去存储一个值的时候，他会根据键值（key）去查看是否存在value值，没有就存储一个值返回true，有就返回一个false，注意一定要加上锁的过期时间，避免线程阻塞。（当用户在请求的过程当中，通过setNX进行加锁完成的时候，这个服务器挂掉了，当其他线程进行setNX进行上锁的时候，发现键当中一直会有值，造成了死锁，其他线程加锁不成功就会造成阻塞）。所以一定要加上过期时间。

        随着业务的扩展，又出现了一些问题，就是1.业务的处理时间超过了锁的过期时间和2.线程1可能释放了线程2所持有的锁，【线程1还没将业务处理完成就释放锁，导致线程2拿到锁处理自己的业务。当线程1执行完成后，释放了锁，但是此时线程2已经拿到了当前锁，所以线程1释放的是线程2的锁。】
        如何解决这些问题呢？
        1.加长锁的过期时间，并增加子线程每10秒去确认线程是否在线。在线则将过期时间重设（续命--他们所说的看门狗）；
        2.给锁的值设置一个唯一ID（UUID）-（使用setNx进行尝试获取锁的时候，如果获取成功，将锁的值设置为一个唯一的ID，释放锁的时候会拿着key去获取锁的值是否与自己的唯一ID一致，一致才进行释放锁，从而就不会释放其他线程的锁）

        上述说这么多，如果让我们自己写起来确实有些麻烦，这时候查阅发现redis他本身已经提供了一个Redission的组件已经解决了这些问题。

那么下面我就提供一个

基于Redission实现分布式锁解决商品秒杀超卖的场景：

springboot版本：2.6.13

redission版本：3.22.0

1.引入依赖：

<dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-data-redis</artifactId></dependency><dependency><groupId>org.redisson</groupId><artifactId>redisson-spring-boot-starter</artifactId><version>3.22.0</version></dependency>

2.加上redis的配置：

server:port: 8083
spring:redis:host: 127.0.0.1password:port: 6379timeout: 1000

3.添加配置类：

import org.redisson.Redisson;
import org.redisson.api.RedissonClient;
import org.redisson.config.Config;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;@Configuration
public class RedissonConfig {@Value("${spring.redis.host}")private String host;@Value("${spring.redis.port}")private String port;@Beanpublic RedissonClient getRedisson(){Config config = new Config();//单机模式  依次设置redis地址和密码config.useSingleServer().setAddress("redis://" + host + ":" + port).setTimeout(30000); // 设置缓存过期时间为30秒return Redisson.create(config);}
}

4.编写代码实现：

import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.web.bind.annotation.GetMapping;import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;import javax.annotation.Resource;
import java.util.Objects;@RestController
@RequestMapping("/redisLock")
public class RedisLockController {@Resourceprivate StringRedisTemplate stringRedisTemplate;@Resourceprivate RedissonClient redisson;// 秒杀商品keyprivate static final String REDIS_KEY = "secKillProductKey:";private static final String LOCK_KEY = "secKillProduct";// 秒杀商品总个数private static final int PRODUCT_SIZE = 1000;/*** 初始化秒杀商品总个数到redis中* 注意：测试的时候，先调用这个接口初始化库存*/@GetMapping("/init")public void init() {// 初始化库存 将库存存到redis中stringRedisTemplate.opsForValue().set(REDIS_KEY, String.valueOf(PRODUCT_SIZE));}/*** 秒杀*/@GetMapping("/secKill")public void secKill() {// 获取锁RLock lock = redisson.getLock(LOCK_KEY);try {// 加锁lock.lock();int s = Integer.parseInt(Objects.requireNonNull(stringRedisTemplate.opsForValue().get(REDIS_KEY)));if (s > 0) {// 扣库存s--;System.out.printf("秒杀商品个数剩余：" + s + "\n");// 更新库存stringRedisTemplate.opsForValue().set(REDIS_KEY, String.valueOf(s));} else {System.out.println("活动太火爆了，商品已经被抢购一空了！");}} catch (Exception e) {System.out.println(Thread.currentThread().getName() + "异常：");e.printStackTrace();} finally {// 释放锁lock.unlock();}}
}

5.模拟服务器分布式集群的情况：

        同一个服务不同端口，同时运行两个相同的主程序。

在service中复制一个进程,指定不同端口

        1.右键点击Copy Configuration

        2.点击Modify option

        3. 选择VM option(用于指定新的端口)

        4.输入想要指定的端口(比如)：-Dserver.port=8082 点击Apply

        5.出现新的进程，点击启动，就可以进行分布式多节点测试。

使用Jmeter进行压测

首先需要先调用一下初始化的接口：127.0.0.1:8082/redisLock/init或127.0.0.1:8083/redisLock/init

1.设置线程组

2.两个HTTP请求，请求不同的端口（8082、8083）进行测试高并发多线程的情况：

3.服务器打印结果

经过测试，一秒钟1000个线程同时请求秒杀接收并没有出现超卖的问题。  

拓展：

       Redis本身是一个CP（一致性和分区容错性）模式的数据库，它通过主从复制实现高可用性，当主节点挂掉时，从节点会自动进行选举，选出一个新的主节点继续提供服务。但是，在主节点挂掉之前，它可能还来不及将最新的数据同步到从节点，这时就会出现数据不一致的问题。

         如果redis采用主从模式进行部署，当往redis中通过setNX进行加锁的过程中，主节点挂了，主节点的数据并没有同步到从节点当中，这种怎么办？

        可以使用RedLock（红锁）解决，RedLock是一个分布式锁的实现，它可以通过访问多个Redis节点来实现更高的可用性和一致性。RedLock的工作原理是，在加锁时，向多个Redis节点发送请求，只有当所有节点都成功返回时，才认为加锁成功。