【Redis】Redis 的学习教程（十一）之使用 Redis 实现分布式锁

2023-09-21 16:54:13

1. 分布式锁概念

在多线程环境下，为了保证数据的线程安全，锁保证同一时刻，只有一个可以访问和更新共享数据。在单机系统我们可以使用 synchronized 锁、Lock 锁保证线程安全。

synchronized 锁是 Java 提供的一种内置锁，在单个 JVM 进程中提供线程之间的锁定机制，控制多线程并发。只适用于单机环境下的并发控制。

想要在多个节点中提供锁定，在分布式系统并发控制共享资源，确保同一时刻只有一个访问可以调用，避免多个调用者竞争调用和数据不一致问题，保证数据的一致性，就需要分布式锁。

分布式锁：控制分布式系统不同进程访问共享资源的一种锁的机制。不同进程之间调用需要保持互斥性，任意时刻，只有一个客户端能持有锁。

共享资源包含：

  • 数据库
  • 文件硬盘
  • 共享内存

分布式锁特性：

  • 互斥性：锁只能被持有的客户端删除，不能被其他客户端删除
  • 锁超时释放：持有锁超时，可以释放，防止不必要的资源浪费，也可以防止死锁
  • 可重入性：一个线程如果获取了锁之后,可以再次对其请求加锁。
  • 高性能和高可用：加锁和解锁需要开销尽可能低，同时也要保证高可用，避免分布式锁失效
  • 安全性：锁只能被持有的客户端删除，不能被其他客户端删除

模拟并发环境下单

①：添加 Redis 依赖：

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

②：添加配置：

spring:
  redis:
    host: localhost
    port: 6379
    password:
    timeout: 2000s
    # 配置文件中添加 lettuce.pool 相关配置，则会使用到lettuce连接池
    lettuce:
      pool:
        max-active: 8  # 连接池最大连接数(使用负值表示没有限制) 默认为8
        max-wait: -1ms # 接池最大阻塞等待时间(使用负值表示没有限制) 默认为-1ms
        max-idle: 8    # 连接池中的最大空闲连接 默认为8
        min-idle: 0    # 连接池中的最小空闲连接 默认为 0

③：Redis 配置类：

@Configuration
public class RedisConfig {

    @Bean
    public RedisTemplate<String, Object> redisTemplate(RedisConnectionFactory redisConnectionFactory) {
        RedisTemplate<String, Object> template = new RedisTemplate<>();
        template.setConnectionFactory(redisConnectionFactory);
        ObjectMapper om = new ObjectMapper();
        om.setVisibility(PropertyAccessor.ALL, JsonAutoDetect.Visibility.ANY);
        om.enableDefaultTyping(ObjectMapper.DefaultTyping.NON_FINAL);
        // json 序列化配置
        Jackson2JsonRedisSerializer<Object> jackson2JsonRedisSerializer = new Jackson2JsonRedisSerializer<>(Object.class);
        jackson2JsonRedisSerializer.setObjectMapper(om);
        // String 序列化
        StringRedisSerializer stringRedisSerializer = new StringRedisSerializer();
        // 所有的 key 采用 string 的序列化
        template.setKeySerializer(stringRedisSerializer);
        // 所有的 value 采用 jackson 的序列化
        template.setValueSerializer(jackson2JsonRedisSerializer);
        // hash 的 key 采用 string 的序列化
        template.setHashKeySerializer(stringRedisSerializer);
        // hash 的 value 采用 jackson 的序列化
        template.setHashValueSerializer(jackson2JsonRedisSerializer);
        template.afterPropertiesSet();
        return template;
    }

}

Redis 工具类：

@Component
public class RedisUtil {

    @Autowired
    private RedisTemplate<String, Object> redisTemplate;

    // 普通缓存获取
    public Object get(String key) {
        return key == null ? null : redisTemplate.opsForValue().get(key);
    }

    // 普通缓存放入
    public boolean set(String key, Object value) {
        try {
            redisTemplate.opsForValue().set(key, value);
            return true;
        } catch (Exception e) {
            e.printStackTrace();
            return false;
        }
    }

    public boolean setIfAbsent(String key, Object value, long timeout, TimeUnit unit) {
        return Boolean.TRUE.equals(redisTemplate.opsForValue().setIfAbsent(key, value, timeout, unit));
    }

    public void del(String... key) {
        if (key != null && key.length > 0) {
            if (key.length == 1) {
                redisTemplate.delete(key[0]);
            } else {
                //springboot2.4后用法
                redisTemplate.delete(Arrays.asList(key));
            }
        }
    }

}

④：添加下单接口

@Slf4j
@RestController
@RequestMapping("/order")
public class OrderController {

    @Autowired
    private RedisUtil redisUtil;

    @GetMapping("/initProductStock")
    public void initProductStock() {
        redisUtil.set("stock", 100);
    }

    @GetMapping("/create_order")
    public void createOrder() {
        // 获取当前库存
        int stock = (Integer) redisUtil.get("stock");
        if (stock > 0) {
            // 减库存
            int realStock = stock - 1;
            redisUtil.set("stock", realStock);
            // TODO 添加订单记录
            log.info("扣减成功，剩余库存：" + realStock);
            return;
        }
        log.error("扣减失败，库存不足");
    }

}

接口说明：

  • /order/initProductStock：先向 Redis 中初始化一个库存
  • /order/create_order：下单接口：先从缓存获取库存，如果库存大于 0，则库存减 1

⑤：并发测试

使用 JMeter 进行并发环境测试，10 个线程，循环 5 次。

在这里插入图片描述

⑥：测试结果，打印日志如下：

在这里插入图片描述
使用 JMeter 调用了 50 次接口后，按照正常情况下，库存应该为：50 = 100 - 50。

但通过日志显示，最终库存为：95。

这是因为在并发环境下，多个线程下单操作，前面的线程还未更新库存，后面的线程已经请求进来，并获取到了未更新的库存，后续扣减库存都不是扣减最近的库存。线程越多，扣减的库存越少。这就是在高并发场景下发生的超卖问题。

很明显，上述问题是出现了线程安全的问题，我们首先能想到的肯定是给它加 synchronized 锁。

是的，没问题，但是我们知道，synchronized 锁是属于JVM 级别的，也就是我们所谓的“单机锁”，如果是多机部署的环境中，还能保证数据的一致性吗？

答案肯定是不能的。这个时候，就需要用到了我们 Redis 分布式锁

用 Redis 实现分布式锁的几种方案，都是用 SETNX 命令（设置 key 等于某 value）。只是高阶方案传的参数个数不一样，以及考虑了异常情况

SETNX 是SET IF NOT EXISTS的简写。命令格式：SETNX key value，如果 key不存在，则 SETNX 成功返回 1，如果这个 key 已经存在了，则返回 0

setIfAbsent() 是 setnx + expire 的合并命令

2. Redis分布式锁方案一：SETNX + EXPIRE

问题：为什么要加过期时间

如果在释放锁之前 Redis 宕机了，就会造成一直死锁。

setnx 命令 和 expire 命令一定要是原子操作。

伪代码如下：

if（jedis.setnx(key_resource_id,lock_value) == 1）{ //加锁
    expire（key_resource_id，100）; //设置过期时间
    try {
        do something  //业务请求
    }catch(){
  }
  finally {
       jedis.del(key_resource_id); //释放锁
    }
}

setnx 和 expire 两个命令分开了，「不是原子操作」。如果执行完 setnx 加锁，正要执行 expire 设置过期时间时，进程 crash 或者要重启维护了，那么这个锁就“长生不老”了，「别的线程永远获取不到锁啦」

@GetMapping("/create_order")
public void createOrder() {
    String key = "lock_key";
    // 1.加锁
    boolean lock = tryLock(key, 1, 60L, TimeUnit.SECONDS);
    if (lock) {
        try {
            // 获取当前库存
            int stock = (Integer) redisUtil.get("stock");
            if (stock > 0) {
                // 减库存
                int realStock = stock - 1;
                redisUtil.set("stock", realStock);
                // TODO 添加订单记录
                log.info("扣减成功，剩余库存：" + realStock);
                return;
            }
            log.error("扣减失败，库存不足");
        } catch (Exception e) {
            log.error("扣减库存失败");
        } finally {
            // 3.解锁
            unlock(key);
        }
    } else {
        log.info("未获取到锁...");
    }
}

public boolean tryLock(String key, Object value, long timeout, TimeUnit unit) {
    return redisUtil.setIfAbsent(key, value, timeout, unit);
}

public void unlock(String key) {
    redisUtil.del(key);
}

使用 JMeter 运行后，结果如下：

在这里插入图片描述

获取到锁的线程已成功扣除库存，没有获取到锁的线程只打印日志。

3. Redis分布式锁方案二：SETNX + EXPIRE + 校验唯一随机值

方案一还是有一定的缺陷的：假设线程 A 获取锁成功，一直在执行业务逻辑，但是 60s 过去了，还没执行完。但是，此时，锁已经过期了。线程 B 又请求过来了，显然，线程 B 也可以获取锁成功，也开始执行业务逻辑代码。那么问题就来了：在线程 B 执行过程中，线程 A 已经执行完了，就会把线程 B 的锁给释放掉！

既然锁可能被别的线程误删，那我们给 value 值设置一个标记当前线程唯一的随机数，在删除的时候，校验一下，就OK了

@GetMapping("/create_order")
public void createOrder() {
    String key = "lock_key";
    String value = "ID_PREFIX" + Thread.currentThread().getId();
    // 1.加锁
    boolean lock = tryLock(key, value, 60L, TimeUnit.SECONDS);
    if (lock) {
        // ...
}

public void unlock(String key, String value) {
    String currentValue = (String)redisUtil.get(key);
    if (StringUtils.hasText(currentValue) && currentValue.equals(value)) {
        redisUtil.del(key);
    }
}

这里需要注意的是：释放锁时，先 get 再删除，这并不是原子操作，无法保证进程安全。为了更严谨，这里用 lua 脚本代替

lua 脚本

Lua脚本是redis已经内置的一种轻量小巧语言，其执行是通过 redis 的 eval/evalsha 命令来运行，把操作封装成一个 Lua 脚本，如论如何都是一次执行的原子操作

lockDel.lua如下：resources/lua/lockDel.lua

if redis.call('get', KEYS[1]) == ARGV[1]
    then
  -- 执行删除操作
        return redis.call('del', KEYS[1])
    else
  -- 不成功，返回0
        return 0
end

public void unlock(String key, String value) {
    // 解锁脚本
    DefaultRedisScript<Long> unlockScript = new DefaultRedisScript<>();
    unlockScript.setScriptSource(new ResourceScriptSource(new ClassPathResource("lua/lockDel.lua")));
    unlockScript.setResultType(Long.class);
    // 执行lua脚本解锁
    Long execute = redisTemplate.execute(unlockScript, Collections.singletonList(key), value);
}

或者：

public void unlock(String key, String value) {
    // 解锁脚本
    String script = "if redis.call('get',KEYS[1]) == ARGV[1] then return redis.call('del',KEYS[1]) else return 0 end";
    redisTemplate.execute(new DefaultRedisScript<Long>(script, Long.class), Collections.singletonList(key), value);
}

4. Redis分布式锁方案三：Redisson

方案二还存在问题：「锁过期释放，业务没执行完」。 如果设置的超时时间比较短，而业务执行的时间比较长。比如超时时间设置5s，而业务执行需要10s，此时业务还未执行完，其他请求就会获取到锁，两个请求同时请求业务数据，不满足分布式锁的互斥性，无法保证线程的安全

4.1 Redisson 概念

其实我们设想一下，是否可以给获得锁的线程，开启一个定时守护线程，每隔一段时间检查锁是否还存在，存在则对锁的过期时间延长，防止锁过期提前释放。当前开源框架 Redisson 解决了这个问题。Redisson 底层原理图如下：

在这里插入图片描述

只要线程一加锁成功，就会启动一个 watch dog 看门狗，它是一个后台线程，会每隔 10 秒检查一下，如果线程 1 还持有锁，那么就会不断的延长锁 key 的生存时间。因此，Redisson 就是使用 watch dog 解决了「锁过期释放，业务没执行完」问题

Redis 虽然作为分布式锁来说，性能是最好的。但是也是最复杂的。上面总结 Redis 主要有下面几个问题：

  • 未设置过期时间，会死锁
  • 设置过期时间
    • 锁误删
    • 业务还继续执行，导致多个线程并发执行

线上都是用 Redission 实现分布式锁，Redisson 是一个在 Redis 的基础上实现的 Java 驻内存数据网格（In-Memory Data Grid）。它不仅提供了一系列的分布式的 Java 常用对象，还提供了许多分布式服务。Redisson 是基于 netty 通信框架实现的，所以支持非阻塞通信，性能优于 Jedis。

Redisson 分布式锁四层保护：

  • 防死锁
  • 防误删
  • 可重入（一个线程可以在获取锁之后再次获取同一个锁，而不需要等待锁释放）
  • 自动续期

Redisson 实现 Redis 分布式锁，支持单机和集群模式

4.2 Redisson 实现

使用 Redission 分布式锁，分成三个步骤：

  1. 获取锁 redissonClient.getLock("lock")
  2. 加锁 rLock.lock()
  3. 解锁 rLock.unlock()

引入依赖：

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

Redisson 配置类：

@Configuration
public class RedissionConfig {

    @Value("${spring.redis.host}")
    private String redisHost;

    @Value("${spring.redis.password}")
    private String password;

    private int port = 6379;

    @Bean
    public RedissonClient getRedisson() {
        Config config = new Config();
        // 单机版
        config.useSingleServer()
                .setAddress("redis://" + redisHost + ":" + port);
        //.setPassword(password);
        config.setCodec(new JsonJacksonCodec());
        return Redisson.create(config);
    }

}

集群版：

@Bean
public RedissonClient getRedisson() {
    Config config = new Config();
    config.useClusterServers()
            .setScanInterval(2000) // 集群状态扫描间隔时间，单位是毫秒
            //可以用"rediss://"来启用SSL连接
            .addNodeAddress("redis://127.0.0.1:7000", "redis://127.0.0.1:7001")
            .addNodeAddress("redis://127.0.0.1:7002");
    return Redisson.create(config);
}

下单接口：

@Slf4j
@RestController
@RequestMapping("/order")
public class OrderController {

    @Autowired
    private RedissonClient redissonClient;

    @GetMapping("/create_order")
    public void createOrder() {
        String key = "lock_key";
        RLock rLock = redissonClient.getLock(key);
        // 1.加锁
        rLock.lock();
        try {
            // 获取当前库存
            int stock = (Integer) redisUtil.get("stock");
            if (stock > 0) {
                // 减库存
                int realStock = stock - 1;
                redisUtil.set("stock", realStock);
                // TODO 添加订单记录
                log.info("扣减成功，剩余库存：" + realStock);
                return;
            }
            log.error("扣减失败，库存不足");
        } catch (Exception e) {
            log.error("扣减库存失败");
        } finally {
            // 3.解锁
            rLock.unlock();
        }
    }
}

使用 JMeter 并发运行后：

在这里插入图片描述

Redission 实现的分布式锁，直接调用，不需要锁异常、超时并发、锁删除等问题，它把处理上面的问题的代码都封装好了，直接调用即可

更多推荐

vue柱状图+折线图组合

<template><divid="main"style="width:100%;height:500px;padding-top:.6rem"></div></template>data(){return{weekData:["1周","2周","3周","4周","5周","6周","7周","8周","9周","

控制台日志打印console的封装，加入美化、行显示与打印开关，支持node.js环境

控制台日志打印console的封装，加入美化、行显示与打印开关，支持node.js环境为什么要写这个？封装这个控制台日志打印工具，主要是在项目中自己做的SDK需要提供给其他开发人员使用，加入了日志美化和打印打开，方便了开发人员查找SDK中的日志（提高了逼格），也加深了自己对日志关键性的理解。首先讲下封装时遇到的一些问题

【力扣】27. 移除元素

题目描述给你一个数组nums和一个值val，你需要原地移除所有数值等于val的元素，并返回移除后数组的新长度。不要使用额外的数组空间，你必须仅使用O(1)额外空间并原地修改输入数组。元素的顺序可以改变。你不需要考虑数组中超出新长度后面的元素。说明:为什么返回数值是整数，但输出的答案是数组呢?请注意，输入数组是以「引用」

ddos打到高防cdn上会发生什么

ddos打到cdn上会发生什么?当DDoS攻击打到CDN上时，肯定会影响网站的可用性和用户体验。具体DDoS攻击打到CDN上时，会发生以下情况：CDN节点负载增加：DDoS攻击会导致大量的无效流量涌入CDN节点，从而使得节点负载增加。这可能会导致节点响应变慢，甚至出现崩溃、停止响应等情况。影响用户访问：如果DDoS攻击

ArcGIS Engine：C#基础语法的了解

目录01前言02实验2.1创建“HelloWorld”程序2.2创建程序，完成成绩评定功能2.3创建程序，完成1到100的累加2.4窗体应用程序01前言学了一些语言了，确实发现，语言只是工具，只有你作为初恋的语言值得你花大量时间去深究。而对于其他作为应用或者说只是作为桥梁的语言去学习其他内容的课程，没有必要过分深究语言

Spring Task

Spring框架提供了一套任务调度的功能，可以帮助开发者实现定时任务和异步任务的管理和调度。任务调度是指按照预定的时间表执行某个任务或一系列任务，Spring的任务调度模块可以在应用程序中方便地配置和管理这些任务。Spring框架中任务调度的核心是通过TaskScheduler接口和@Scheduled注解来定义和执行

OpenHarmony ArkTS工程目录结构（Stage模型）

一、应用工程结构图片来源：OpenHarmony官网AppScope>app.json5：应用的全局配置信息。entry：OpenHarmony工程模块，编译构建生成一个HAP包。src>main>ets：用于存放ArkTS源码。src>main>ets>entryability：应用/服务的入口。src>main>e

Java拦截器与过滤器的区别

主要结论：运行顺序不同，过滤器先，拦截器后配置方式不同，过滤器web.xml，拦截器spring的配置文件过滤器依赖于servlet，拦截器依赖于Spring过滤器只能对request和response响应，拦截器还能对springmvc生态下的组件做处理。（说人话就是咱们现在用的都是人家spring的产品，那么拦截器

golang入门笔记——pprof性能分析

文章目录简介runtime/pprof的使用命令行交互网络服务性能分析pprof与性能测试结合压测工具go-wrk简介golang性能分析工具pprof的8个指标1.性能分析的5个方面：CPU、内存、I/O、goroutine（协程使用情况和泄漏检查）、死锁检测以及数据竟态分析runtime.SetMutexProfi

天地图绘制区域图层

背景：业务方要求将原效果图参考效果图最终实现效果变更点：1.将原有的高德地图改为天地图2.呈现形式修改：加两层遮罩：半透明遮罩层mask+区域覆盖物mask实现过程：1.更换地图引入源<linkrel="stylesheet"href="https://cdn.jsdelivr.net/npm/maptalks/dis

IntelliJ IDEA使用——常规设置

文章目录版本说明主题设置取消检查更新依赖自动导入禁止importxxx.*、允许import内部类显示行号、方法分割线、空格代码提示（匹配所有字母）自定义注释颜色添加头部注释自定义字体设置字符编码关联本地GitJDK编译版本Maven配置Tomcat配置代码注释设置头部注释单行注释HTML和XML注释IDEA同步设置版

热文推荐
  1. 解决Nacos配置刷新导致定时器停止执行的问题

    1.问题描述我使用了一个定时器类来执行某个任务，并且使用Nacos作为配置中心来管理定时器的配置。我发现当Nacos配置发生变化时，定时器实例会停止执行任务，导致任务无法按预期执行。2.原先的实现方式以下是我原先的代码实现方式：@Component@RefreshScope@RequiredArgsConstructo

  2. 英飞凌TC3xx--深度手撕HSM安全启动(四)--TC3xx HSM使能和配置技巧

    上一章，我们简单聊了下英飞凌TC3xx的HSM的系统框架、相关UCB、Host和HSM通信模块。今天着重分析HSM的使能。1.系统引入HSM的思考为什么要增加HSM信息安全方面考虑，系统的安全启动、ECU之间安全数据的交互、ECU内部的敏感信息保存TC3xx使能HSM后，HSM的代码应该存放在哪里？在上一章，我们了解到

  3. C++的移动构造和移动赋值运算符

    右值引用右值引用(rvaluereferences)是一种新的用于绑定右值的引用类型。那么什么是右值？右值通常是编译器生成的用于表达式计算的临时变量或常量。目前来说，我们还不能安全地使用引用变量来绑定右值。从编译原理上讲，右值是只存在于表达式计算时的未命名值。下面这一表达式产生了一个右值：x+(y*z);//AC++e

  4. SpringBoot携带Jre绿色部署项目_免安装Jdk[Linux服务器]

    文章目录SpringBoot携带Jre绿色部署项目[Linux服务器]1.实现步骤2.自测成功，如下2-1环境准备2-2运行项目SpringBoot携带Jre绿色部署项目[Linux服务器]说明：实际应用的不方便场景：1.实际项目部属时，现有服务器可能已安装有Jdk，和自己项目的Jdk版本不一致，不敢轻易安装自己使用的

  5. 解决Permission is not allowed后基于Ubuntu23.04安装配置docker与docker-compose

    参考：Docker官网-InstallDockerEngineonUbuntu虚拟机里安装ubuntu-23.04-beta-desktop-amd64，开启SSH(换源、备份)，配置中文以及中文输入法等基于CentOS7安装配置docker与docker-compose一、InstallusingtheAptrepo

  6. 代码随想录算法训练营第二天(C) | 977.有序数组的平方 209.长度最小的子数组 59.螺旋矩阵

    文章目录前言一、977.有序数组的平方二、209.长度最小的子数组三、59.螺旋矩阵总结前言java版：代码随想录算法训练营第二天|977.有序数组的平方，209.长度最小的子数组，59.螺旋矩阵_愚者__的博客-CSDN博客一、977.有序数组的平方双指针法：int*sortedSquares(int*nums,in

  7. 二进制 Deploy Kubernetes v1.23.17 超级详细部署

    文章目录1.预备条件2.基础配置2.1配置root远程登录2.2配置主机名2.3安装ansible2.4配置互信2.5配置hosts文件2.6关闭防firewalld火墙2.7关闭selinux2.8关闭交换分区swap2.9修改内核参数2.10安装iptables2.11开启ipvs2.12配置limits参数2.1

  8. nginx详解

    目录1什么是nginx2Nginx功能2.1正向代理2.2反向代理2.3负载均衡2.4动静分离3区别4Docker安装nginx5nginx配置文件介绍1什么是nginxnginx是一个高性能的http和反向代理服务器，特点是占有内存少、并发能力强，事实上nginx的并发能力的确在同类的服务器中表现较好。nginx转为

  9. 聊一聊Twitter的雪花算法

    什么是Twitter的雪花算法方法？这是一种在分布式系统中生成唯一ID的解决方案。Twitter在推文、私信、列表等方面使用这种方法。•ID是唯一且可排序的•ID包含时间信息（按日期排序）•ID适用于64位无符号整数•仅包含数字值符号位（1位）：保留位（始终为0）。这可以为将来的请求保留。它可以潜在地用于使整体数字为正