Redis 缓存与分布式锁

发表于 更新于 阅读次数： 评论数： 本文字数： 4.9k 阅读时长 ≈ 4 分钟

缓存

缓存使用

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

哪些数据适合放入缓存？

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

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

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

特别注意

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

SpringBoot 整合 Redis

  • Maven 引入 SpringBoot 的 Redis Starter
1
2
3
4
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-data-redis</artifactId>
</dependency>

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

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
<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 配置文件中添加以下内容
1
2
3
4
5
spring:
  redis:
    port: 6379
    host: 192.168.56.103
    password: 123456
  • 简单使用 RedisTemplate 类操作 Redis
1
2
3
4
5
6
7
8
9
10
@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 文件，用于保证解锁操作的原子性
1
2
3
4
5
6
if redis.call("get", KEYS[1]) == ARGV[1]
then
    return redis.call("del", KEYS[1])
else
    return 0
end

提示

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

  • 引入 Maven 坐标
1
2
3
4
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-data-redis</artifactId>
</dependency>
  • Lua 脚本的配置类
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
@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 分布式锁的服务类
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
@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);
    }
    
}
  • 单元测试
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
@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();
    }
    
}
  • 测试结果
1
2
3
4
5
6
7
8
9
10
11
==> 加锁成功
==> 加锁失败
==> 加锁失败
==> 加锁失败
==> 加锁失败
==> 加锁失败
==> 加锁失败
==> 加锁失败
==> 加锁失败
==> 加锁失败
==> 解锁成功

总结

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

Redisson 分布式锁的使用

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

Redisson 基础使用教程

参考博客