Redis 缓存与分布式锁

发表于 2021-02-08 更新于 2024-05-21 阅读次数：评论数：本文字数： 4.9k 阅读时长 ≈ 4 分钟

缓存

缓存使用

为了系统性能的提升，一般都会将部分数据放入缓存中，加快业务服务的处理速度，而数据库则承担数据落盘的工作。

哪些数据适合放入缓存？

即时性、数据一致性要求不高的数据
访问量大且更新频率不高的数据（读多写少）

比如在电商类应用中，商品分类，商品列表等数据适合缓存，并加一个失效时间 (根据数据更新频率来决定)，后台如果发布一个商品，买家需要 5 分钟后才能看到新的商品，这一般还是可以接受的。

缓存读模式的使用流程图如下

特别注意

在开发中，凡是放入缓存中的数据都应该指定过期时间，使其可以在系统即使没有主动更新数据的情况下，也能自动触发数据加载进缓存的流程。避免出现业务崩溃导致的数据永久不一致问题。

SpringBoot 整合 Redis

Maven 引入 SpringBoot 的 Redis Starter

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

SpringBoot 2.0 以后默认使用的 Redis 客户端是 Lettuce，若希望使用 Jedis 作为客户端（不推荐），可以使用以下 Maven 配置信息

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-data-redis</artifactId>
    <exclusions>
        <exclusion>
            <groupId>io.lettuce</groupId>
            <artifactId>lettuce-core</artifactId>
        </exclusion>
    </exclusions>
</dependency>

<dependency>
    <groupId>redis.clients</groupId>
    <artifactId>jedis</artifactId>
</dependency>

堆外内存溢出

对 Lettuce 进行压力测试时，可能会出现 io.netty.util.internal.OutOfDirectMemoryError 的错误，即产生了堆外内存溢出。主要原因是 SpringBoot 2.0 以后默认使用 Lettuce 作为操作 Redis 的客户端，它是基于 Netty 进行网络通信，而由于旧版 Lettuce 自身的 Bug 导致一些使用过的内存没有被及时清理掉，因此最终会出现内存溢出的问题。解决方案有两种：一是升级 Lettuce 的版本，而是使用 Jedis 作为 Redis 的客户端。

配置 Redis 的连接信息，在 application.yml 配置文件中添加以下内容

spring:
  redis:
    port: 6379
    host: 192.168.56.103
    password: 123456

简单使用 RedisTemplate 类操作 Redis

@Autowired
public StringRedisTemplate stringRedisTemplate;

@Test
public void testStringRedisTemplate() {
    ValueOperations<String, String> ops = stringRedisTemplate.opsForValue();
    ops.set("hello", "world_" + UUID.randomUUID().toString());
    String hello = ops.get("hello");
    System.out.println(hello);
}

缓存失效问题

在高并发的业务场景下，缓存失效一般分为几种情况，包括 缓存穿透、缓存击穿、缓存雪崩。

缓存穿透

缓存穿透是指查询一个不存在的数据。由于数据不存在，缓存没有命中，将去查询数据库，但是数据库也无此记录，因此没有将这次查询的 Null 结果写入缓存；这将导致这个不存在的数据每次请求都要到数据库去查询，失去了缓存的意义。在流量大的时候，数据库可能就会被压垮，要是有人恶意利用不存在的 Key 频繁攻击应用服务，这就存在安全漏洞。为了解决缓存穿透的问题，可以采用以下方案：

对用户输入的参数进行严格校验，过滤掉参数非法或参数明显不存在的请求，减少对缓存和数据库的压力。
当数据库查询不到数据时，可以将空值（Null）写入缓存，并设置较短的过期时间（防止空值占用大量内存空间）。
引入布隆过滤器，在访问 Redis 之前判断数据是否存在，如果不存在就直接返回，不再访问数据库。特别注意，布隆过滤器存在一定的误判率，并且布隆过滤器只能加数据不能减数据。

布隆过滤器的特性

如果布隆过滤器判断元素不在一个集合中，那么元素就肯定不在集合当中。
如果布隆过滤器判断元素在一个集合中，这时会存在一定的误判率，也就是元素不一定真的在集合当中。

缓存击穿

缓存击穿是指热点数据过期，导致大量外部请求转发到数据库。对于一些设置了过期时间的 Key，如果这些 Key 可能会在某些时间点被超高并发地访问，也就是一种非常 热点 的数据。那么在这个时候，需要考虑一个问题，如果这个 Key 在大量请求同时进来前刚好失效，那么所有对这个 Key 的数据查询都会落到数据库，这种现象一般称为 缓存击穿。为了解决缓存击穿的问题，可以采用以下方案：

可以使用 Cananl 数据库中间件
当缓存没有命中时，可以加锁（分布式锁 + DCL 双端检测）来限制对数据库的并发访问。
设置缓存中的热点数据永不过期，这时候需要注意在缓存数据中包含一个逻辑上的过期时间属性，然后后端另起一个线程，定期更新这些缓存（只能保证缓存数据的最终一致性）。

缓存雪崩

缓存雪崩是指在设置缓存时，采用了相同的过期时间，导致大量缓存在某一时刻同时失效，外部请求全部转发到数据库，而数据库由于瞬时压力过重导致雪崩。为了解决缓存雪崩问题，可以采用以下方案：

可以在原有的缓存失效时间基础上增加一个随机值，比如 1 ~ 5 分钟随机，这样每一个缓存的过期时间的重复率就会降低，进而很难引发缓存集体失效的事件。

缓存更新问题

更新缓存数据时，一般有两种模式（统称写模式），分别是 双写模式 与 失效模式，这两种写模式都存在缓存数据一致性问题（即可能会读取到脏数据）。

双写模式

双写模式 是指先将数据写入数据库，然后再写入缓存。在 双写模式 下，读到的数据可能会不是最新的（存在延迟），同时还可能会读取到暂时性的脏数据，图解说明如下。值得一提的是，双写模式 属于 最终一致性 的一类。

脏数据问题分析

如上图，线程 A 和 B 都去写数据库，正常情况下应该是，A 先写数据库先写缓存，B 后写数据库后写缓存；但是由于卡顿等原因（如网络卡顿），导致写缓存 2 在最前，写缓存 1 在后面，这就会造成缓存数据与数据库数据不一致，即可能会读取到脏数据；但是这是暂时性的脏数据问题，在数据稳定和缓存过期以后，又能得到最新的正确数据。若希望从根本上解决脏数据的问题，可以使用分布式锁（读写锁），也就是写数据库和写缓存这两个操作（两者可以是看做是一个操作）需要获取到锁才能执行，但是加了分布式锁以后，系统的整体性能会下降。另外，也可以使用 MQ 或者 Canal 中间件来解决缓存的一致性问题。

失效模式

失效模式 是指写完数据库，不用写缓存，而是删除缓存；等有请求进来读数据的时候，发现缓存中没有数据，就会主动查询数据库，并将查询结果放到缓存里面，这也叫 触发主动更新。值得一提的是，失效模式 也存在读取到脏数据的问题，如下图所示。

缓存击穿问题

失效模式 会导致缓存击穿问题产生。因为当缓存数据被删除后，可能会导致大量外部请求转发到数据库，所以在更新缓存数据（查询数据库并写入缓存）的过程中，需要进行同步控制（分布式锁），即同一时间只允许有一个请求访问数据库。

缓存一致性解决方案总结

无论是双写模式还是失效模式，都会导致数据库数据与缓存数据不一致的问题，即多个实例同时更新时会出事，那么怎么办呢？

缓存数据 + 过期时间的配合使用，也足够解决大部分业务对于缓存的要求。
如果是用户维度的数据 (订单数据、用户数据)，这种数据并发更新的几率非常小，可以不用考虑一致性问题，缓存数据加上过期时间，每隔一段时间自动触发读的主动更新即可。
如果是菜单列表、商品介绍等基础数据，也可以使用 Canal 中间件订阅数据库 binlog 的方式来更新缓存。
针对 失效模式 的缓存击穿问题，需要使用分布式锁进行同步控制，也就是在更新缓存数据（查询数据库并写入缓存）的过程中，保证同一时间只允许有一个请求访问数据库。
针对 双写模式 的缓存数据一致性问题，可以通过使用分布式锁来保证并发读写的准确性，写 + 写 的时候按顺序排好队执行，读 + 读 则无所谓，所以适合使用分布式读写锁（如果业务不关心脏数据，允许临时的脏数据存在，则可以不使用分布式锁）。

解决缓存数据一致性问题，有以下几种方案：

方案的名称	方案的类型	方案的一致性	方案的实现思路	备注
缓存数据实时同步更新	增量 / 主动更新	最终一致性	更新数据库数据之后，立刻更新缓存数据；为了尽量避免缓存数据一致性问题，需要使用分布式锁（读写锁），也就是更新数据库和更新缓存这两个操作（两者可以是看做是一个操作）需要获取到锁才能执行。存在缓存数据一致性问题（即可能会读取到脏数据）。	双写模式
缓存数据实时同步失效	增量 / 主动更新	最终一致性	更新数据库数据之后，立刻删除缓存数据，读请求会更新缓存数据；为了避免缓存击穿问题，在更新缓存数据的过程中需要进行同步控制（分布式锁），即同一时间只允许有一个请求访问数据库。存在缓存数据一致性问题（即可能会读取到脏数据）。	失效模式
缓存数据准实时更新	增量 / 主动更新	最终一致性	更新数据库数据之后，异步更新缓存数据，可以使用 MQ 或者 Canals 实现缓存数据的异步更新。存在缓存数据一致性问题（即可能会读取到脏数据）。
任务调度更新缓存数据	全量 / 被动更新	最终一致性	采用任务调度框架，按照一定的频率定时更新缓存数据。存在缓存数据一致性问题（即可能会读取到脏数据）。

总结

能放入缓存的数据本就不应该是实时性、一致性要求超高的，所以一般在缓存数据的时候加上过期时间，保证每天拿到当前最新的数据即可。
遇到实时性、一致性要求高的数据，就应该直接查询数据库，即使效率慢一点。
系统不应该过度设计，否则会增加系统的复杂性。

使用 MQ 解决一致性问题

使用 MQ 消息队列中间件，可以从根本上解决缓存一致性的问题，但会增加系统的复杂性，整理的工作流程图如下：

使用 Canal 解决一致性问题

使用 Canal 数据库中间件，可以从根本上解决缓存一致性的问题，但会增加系统的复杂性，整理的工作流程图如下：

本地锁与分布式锁

本地锁

本地锁，如使用 JDK 的 synchronized 关键字或者 JUC 包下的 Lock 类等。本地锁只能锁住当前的 Java 进程，并不适用于分布式的业务场景。

注意

使用本地锁操作缓存时，需要注意锁的时序问题，即查询数据库与写入缓存这两者必须是原子操作，点击查看详细的图解说明。

分布式锁

分布式锁，如使用 Redisson 第三方库提供的各种锁。分布式锁可以简单理解为同时去一个地方 占坑，如果占到，就执行业务逻辑，否则就必须等待，直到占到锁为止。占坑 可以去 Redis，也可以去数据库。等待过程可以是使用自旋的方式。

Redis 分布式锁的实现

这里将介绍如何使用 Redis 实现分布式锁，更多内容建议参考 Reids 官方中文文档。

实现命令

set resource-key resource-value nx ex max-lock-time：设置 key 和设置过期时间，属于原子命令
例如： set sku 56a4e5e-a022 nx ex 300，其中的 key 是 sku，value 是 56a4e5e-a022，过期时间是 300 （单位是秒），设置成功会返回 OK，否则返回 Nil

实现流程

核心问题

使用 Redis 实现分布式锁，最重要的是锁要有过期时间，不然万一业务代码抛出异常或者 Redis 宕机，Redis 锁将永远得不到释放，进而出现 死锁，导致其他线程一直获取不到资源。为了避免这种情况的发生，就必须保证执行加锁时，设置 Key 与设置过期时间这两个操作的原子性（不可分割）。值得一提的是，在解锁的时候，也必须保证判断 Key 的值（锁的归属）和删除 Key 这两个操作的原子性（不可分割）。

Redis 实现分布式锁的核心内容

保证加锁操作的原子性：通过 Redis 自身的 SET key value EX seconds NX 命令加锁。
保证解锁操作的原子性：通过 Lua 脚本实现解锁，不能直接使用 DEL 命令删除锁。
在执行解锁时，必须确保解锁的是自己加的锁。

代码实现

在项目的 resources 目录下创建 lua 目录，并且创建 redisLock.lua 文件，用于保证解锁操作的原子性

if redis.call("get", KEYS[1]) == ARGV[1]
then
    return redis.call("del", KEYS[1])
else
    return 0
end

提示

更多关于 Redis 分布式锁中 Lua 脚本的使用教程，请点击这里。

引入 Maven 坐标

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

Lua 脚本的配置类

@Configuration
public class RedisLuaConfig {
    
    @Resource
    private StringRedisTemplate stringRedisTemplate;
    
    /**
     * 删除锁
     */
    public boolean deleteLock(String lockKey, String value) {
        List<String> keyList = Collections.singletonList(lockKey);
        DefaultRedisScript<Long> redisScript = new DefaultRedisScript<>();
        redisScript.setScriptSource(new ResourceScriptSource(new ClassPathResource("lua/redisLock.lua")));
        redisScript.setResultType(Long.class);
        Long result = stringRedisTemplate.execute(redisScript, keyList, value);
        return 1 == result;
    }
    
}

Redis 分布式锁的服务类

@Service
public class RedisLockService {
    
    @Resource
    private RedisLuaConfig redisLuaConfig;
    
    @Resource
    private StringRedisTemplate stringRedisTemplate;
    
    /**
     * 加锁（原子操作）
     */
    public boolean lock(String lockKey, String value, long time, TimeUnit timeUnit) {
        return stringRedisTemplate.opsForValue().setIfAbsent(lockKey, value, time, timeUnit);
    }
    
    /**
     * 解锁（原子操作）
     */
    public boolean unlock(String lockKey, String value) {
        return redisLuaConfig.deleteLock(lockKey, value);
    }
    
}

单元测试

@SpringBootTest
public class RedisLockTest {
    
    @Autowired
    private RedisLockService lockService;
    
    /**
     * 加锁/解锁
     */
    private void reduceSku() throws Exception {
        // 锁的唯一标识，用于保证解锁的是当前线程自己加的锁
        String value = UUID.randomUUID().toString();
        
        // 加锁 + 设置锁的过期时间，必须是原子操作
        boolean lock = lockService.lock("lock", value, 10, TimeUnit.SECONDS);
        if (lock) {
            System.out.println("==> 加锁成功");
            
            // 模拟业务执行的耗时
            TimeUnit.SECONDS.sleep(8);
            
            // 解锁，必须满足原子性，通过 Redis + Lua 脚本实现
            boolean unlock = lockService.unlock("lock", value);
            System.out.println("==> 解锁" + (unlock ? "成功": "失败"));
        } else {
            System.out.println("==> 加锁失败");
        }
    }
    
    /**
     * 并发测试
     */
    @Test
    public void multiThreadLock() throws Exception {
        for (int i = 0; i < 10; i++) {
            new Thread(() -> {
                try {
                    reduceSku();
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }).start();
        }
        System.in.read();
    }
    
}

测试结果

==> 加锁成功
==> 加锁失败
==> 加锁失败
==> 加锁失败
==> 加锁失败
==> 加锁失败
==> 加锁失败
==> 加锁失败
==> 加锁失败
==> 加锁失败
==> 解锁成功

总结

为了防止持有过期锁的客户端误删现有锁的情况出现，可以使用以下方案改进
a) 不使用固定的字符串作为键的值，而是设置一个不可猜测（如 UUID）的长随机字符串作为口令串（token）。
b) 不使用 DEL 命令来解锁，而是发送一个 Lua 脚本，这个脚本只在客户端传入的值与口令串相匹配时，才对键进行删除。

Redisson 分布式锁的使用

上面介绍的方式并不推荐用来实现 Redis 分布式锁。Redis 官方推荐参考 the Redlock algorithm 的实现，因为这种方法只是复杂一点，但是却能保证更好的使用效果。其中，基于 Java 语言开发的分布式锁的框架就是 Redisson。

Redisson 基础使用教程

Redisson 的使用请阅读 Redisson 分布式锁使用教程。

缓存

缓存使用

SpringBoot 整合 Redis

缓存失效问题

缓存穿透

缓存击穿

缓存雪崩

缓存更新问题

双写模式

失效模式

缓存一致性解决方案总结

使用 MQ 解决一致性问题

使用 Canal 解决一致性问题

本地锁与分布式锁

本地锁

分布式锁

Redis 分布式锁的实现

实现命令

实现流程

核心问题

代码实现

Redisson 分布式锁的使用

参考博客