为什么需要多级缓存

缓存的意义

缓存是数据库数据的临时仓库，更是系统性能的"加速器"。

如果没有缓存，每次取数据，都需要从数据库中拿，会产生以下成本：

1. 网络成本：通过 TCP/IP 协议发送请求、接收响应，Java对象数据序列化
2. 硬件资源成本：CPU 解析执行 SQL，进行磁盘 IO，占用大量内存带宽进行数据读写
3. 并发成本：为保证数据一致性，对读写加锁，有锁就有等待成本；数据库连接池容量也是有限的，大量请求会因等待连接而超时失败

使用缓存之后就会先从缓存（Caffeine 或 Redis）中查询，会相应带来以下好处：

1. 网络：由于缓存数据存储在内存中，且本地缓存无需网络传输，查询速度会非常快
2. 硬件资源：请求被缓存拦截后，数据库读写压力大幅下降，CPU、磁盘 IO 和网络带宽消耗显著减少
3. 并发：缓存能分流大部分查询请求，减少数据库的锁竞争和连接池压力

本地缓存 Caffeine

Caffeine 是 Java 中i性能最优的本地缓存，其实可以将其当作一个HashMap，只不过是在 HashMap 上，添加了一些功能：

1. 并发安全：无锁读（通过 volatile 保证可见）；细粒度桶锁写，仅对目标数据所在桶加锁，冲突概率极低
2. 缓存淘汰算法：W-TinyLFU 算法，比 LRU（最近最少）算法命中率高

▼
plain
复制代码
1. 新来的先放临时区（Window Cache），这里专门放新面孔，如果临时区满了，就把最久没人碰过的东西挪到观察区

2. 观察区（Probation Cache），如果观察区没满，就直接进去，如果满了，就和观察区最久没碰过的 "老东西" 比 "人气" 
- 新来的人气高，就把老东西扔了
- 老东西人气高，如果新来的人气太低（少于5次访问），就直接扔了；不然就随机扔一个

3. 保护区（Protected Cache），如果观察区里面的东西被频繁访问（达到一定次数），就升级到保护区，这里的东西不容易被扔，如果保护区满了，就把最久没人碰过的东西挪回观察区，再按照上面的规则判断留不留

总结：LRU 算法只看最近的访问时间，如果某个数据被长期访问，但某天被临时数据挤到后面去，会被误删；但是本算法会统计访问频率，通过时间和频率双重判断，新数据先观察，高频数据先保护，低频数据才被淘汰，本算法思想适用于资源有限但需高效利用的场景，例如抖音短视频推荐、淘宝商品推荐，特此记录

3. 灵活配置：可自主配置过期策略；控制缓存的容量（数量 或者 占用内存的权重）；自动加载（缓存未命中时，自动执行预设逻辑，查询Redis 或者数据库，并回填缓存，但是这个我感觉不好用，我一般自己写逻辑）

分布式缓存 Redis

本地缓存（Caffeine）存在一些缺点：

1. 节点私有，比如有10台服务器，每台服务器上的本地缓存都是独立的，如果某台服务器更新了数据，其它9台无法同步数据。
2. 内存限制，每台服务器缓存有限，无法存储海量的
3. 断电清空，它是随应用进程存在，一旦应用重启或崩溃，缓存数据会全部丢失

由此 Redis 闪亮登场：

1. 节点公有：所有服务器访问同一份 Redis 缓存，数据一旦更新，所有节点查询都能拿到最新值
2. 内存无限：通过集群配置，将数据分散到多态服务器，理论上可无限扩容
3. 持久化+高可用：通过 RDB（定期对数据库拍快照） 和 AOF（实时记录所有写操作） 持久化，即使 Redis 重启，也能通过持久化文件恢复数据，同时还支持主从复制、哨兵模式、集群模式，自动切换主从节点，保证缓存服务不中断

相互配合

流程：

1. Caffeine 命中时，直接返回数据；
2. Caffeine 未命中时，会先查询Reids，如果 Redis 有数据则回写到 Caffeine 并返回数据；
3. Reids 没有数据，才去查数据库，然后写到 Redis，再写到 Caffeine ，再返回数据

虽然 Redis 访问需要网络传输，但是通过合理设计（如 Caffeine 缓存高频数据，Redis 存储需要共享的数据），可让 Redis 专注于"一致性保障"，而 Caffeine 专注于 "性能加速"。

多级缓存的核心挑战与应对策略

缓存穿透

当查询一个不存在的数据（如id = -1），会穿透 Caffeine 和 Redis，直接打到数据库，高并发下可能会拖垮数据库。

解决方案：

1. 合法性检查：先校验查询条件是否合法，非法直接拦截
2. 缓存空值：当数据库查询为空时，在 Caffeine 和 Redis 中缓存 "空值"，并设置较短的过期时间如（Caffeine 1分钟， Redis 5分钟），后续相同的查询会被拦截，不再访问数据库
3. 布隆过滤器（海量数据场景）：在 Redis 前加一层布隆过滤器，提前判断 "查询的key是否在数据库中"（概率判断，误判率极低）

▼
latex
复制代码
1. 初始化：准备一个长度为 m 的二进制数组，均初始化为 0 ，以及 k 个不同的 哈希函数（如哈希函数 A、B、C）

2. 添加元素，对元素用上面的 k 个哈希函数进行哈希，得到 k 个不同的数组下标（如 A 算到 5，B 算到12，C 算到 23），将这几个下标从 0 改成 1

3. 判断元素是否存在，同样使用这 k 个哈希函数进行哈希，如果所有的下标查出来都是 1，那可能存在（允许误判），如果任何一个下标不是 1，那一定不存在

简单举例：
假设有个书架，上面有 1000 个格子，每个格子都贴一个 有/无 标签，要用三个哈希函数来贴标签，
比如要登记西游记、红楼梦，分别计算，得到100，200，300；150，250，350，都贴上有，其它都是无
现在我要查三国演义，计算出来是120，200，300，有一个格子不是，那代表没有，或者碰巧了，这仨恰好都是有，那允许偶尔认错

缓存击穿

当某条高频访问的数据，同时在 Caffeine 和 Redis 过期，大量的并发请求会打到数据库

解决方案：

1. 分布式锁+缓存重建：缓存未命中时，只有一个线程可以获取到分布式锁，其它线程等着，拿锁的线程查到数据重新构建缓存，其它线程从缓存中获取结果
2. 热点数据永不过期+主动更新：对核心热点数据，Caffeine 和 Redis 均不设置过期时间，数据更新通过mq通知更新缓存
3. 逻辑过期+被动更新：不设置过期时间，会导致数据泄露（比如活动过期了，数据也应该过期），在缓存结构中设置逻辑过期时间，到时间后，再次请求，返回原来的旧数据，相数据库发起查询请求，或者进行其它的处理

缓存雪崩

出现原因：

1. 大量 key 集中过期：若缓存中的大量 key 设置了相同的过期时间（如凌晨 3 点统一过期），到时间后所有请求会涌向数据库
2. 缓存服务故障：Reids 集群宕机、网络波动等导致缓存整体不可用，所有请求直接穿透到数据库

解决方案：

1. 过期时间加随机值：在基础过期时间上增加 0-300 秒的随机值，多级缓存过期时间错开
2. 缓存集群化部署：采用主从、哨兵、集群模式部署；采用限流、熔断降级策略

数据一致性

当数据发生更新（如商品价格调整），若只更新数据库，未即时同步到 Redis 或 Caffeine，会导致仍然返回旧数据（也就是脏读）

首先说明，没有解决数据一致性问题的 "银弹"，只有最适合业务的解决方案

要一致性，就需要考虑到删除缓存，于是就有两种方式

1. 先删除缓存，再更新数据库
2. 先更新数据库，再删除缓存

我们先看第一种方式，在高并发情况下，假如删除缓存之后，还没来得及更新数据库或者说更新数据库失败了，另一个线程发现没有缓存，进而又去读取数据库，并且回写缓存，就会导致数据不一致。

再看第二种方式，在高并发下，假如更新数据库之后，还没来得及删除缓存或者说删除缓存失败，另一个线程就会读取旧缓存，导致数据不一致

一般情况下选择第二种方式，因为第二种方式，我们可以使用其它的方法来减轻或减缓其问题。

下面均采用这种方式来解决其问题。

解决方案：

1. 延时双删：修改数据库后，先删除 Caffeine，再删除 Redis 缓存，然后设置异步延时再次进行删除Redis 缓存，由于 Caffeine 存在于本进程中，并不是分布式的，且删除 Caffeine 失败概率极小，因此无需延时再次删除本地缓存；

2. 失败重试：但是此时删除缓存可能还是会失败，可以将异步删除放入消息队列中，或者如果失败再放入消息队列中，在消息队列中可以进行重试，以确保最终将缓存删除，Caffeine 在 1 中，已经说明失败概率极小，如果失败了，可能是代码本身有问题，可直接抛异常报错；

3. 订阅变更日志：加入消息队列会产生额外的维护成本，此时可以使用 canal，当成功修改 MySQL 之后，会产生 Binlog，canal 会自动将日志投递到下游的消息队列中

4. 分布式锁：当然，上面所述方案还是不能达到完全一致性，那我们可以为其加入分布式锁，可是一旦加上分布式锁，代表性能会下降，那我们使用缓存的意义又是什么呢，因此不建议采用分布式锁。
5. 过期兜底：虽然上面的方案已经做了很多事情了，但是无论如何，我们都应该为缓存加上过期时间，且本地缓存 < Redis 缓存达到兜底方案，避免中间服务任何一步出现错误，来让数据长期不一致

实战（基础）

背景：查询商品

开源地址：https://github.com/sonshu-tulin/multi-cache

引入依赖

▼
xml
复制代码
<!-- Redis -->
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-data-redis</artifactId>
</dependency>

<!-- 本地缓存 Caffeine -->
<dependency>
    <groupId>com.github.ben-manes.caffeine</groupId>
    <artifactId>caffeine</artifactId>
    <version>3.1.8</version>
</dependency>

创建表

▼
sql
复制代码
-- 商品表：存储商品基本信息（ID、名称、价格、库存等）
CREATE TABLE IF NOT EXISTS product (
   id BIGINT UNSIGNED NOT NULL AUTO_INCREMENT COMMENT '商品ID（主键）',
   name VARCHAR(255) NOT NULL COMMENT '商品名称',
   price DECIMAL(10, 2) NOT NULL COMMENT '商品售价（精确到分）',
   stock INT NOT NULL DEFAULT 0 COMMENT '商品库存数量',
   update_time BIGINT NOT NULL COMMENT '最后更新时间（时间戳，用于版本控制）',
   create_time TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',
   PRIMARY KEY (id),
   KEY idx_update_time (update_time) COMMENT '索引：按更新时间查询（优化缓存同步场景）'
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='商品信息表';

使用Mybatis-X生成代码结构

在表右键点击MybatisX插件生成

选择目录和表

选择mybatisplus3

这是生成之后的项目结构

代码实现

1. 写 Controller 结构

▼
java
复制代码
@RestController
@RequestMapping("/products")
public class ProductController {

    @Resource
    private ProductService productService;

    // 查询商品详情
    @GetMapping("/{id}")
    public Product getProduct(@PathVariable Long id) {
        return productService.getProductDetail(id);
    }
}

2. 写 Service 实现

▼
java
复制代码
public interface ProductService extends IService<Product> {
    Product getProductDetail(Long id);
}

3. 写 Impl 实现

可以分为 5 步：

▼
java
复制代码
@Service
public class ProductServiceImpl extends ServiceImpl<ProductMapper, Product> implements ProductService {

    // 1. 创建缓存
    // 引入 Redis
    @Resource
    private StringRedisTemplate stringRedisTemplate;

    // 创建本地缓存
    public final Cache<String, String> LOCAL_CACHE =
            Caffeine.newBuilder().initialCapacity(1024)
                    .maximumSize(10000L)
                    // 缓存 5 分钟移除
                    .expireAfterWrite(5L, TimeUnit.MINUTES)
                    .build();


    @Override
    public Product getProductDetail(Long id) {
        // 2. 构建缓存key
        String redisKey = DigestUtils.md5DigestAsHex(id.toString().getBytes());
        String cacheKey = "cache" + redisKey;

        // 3. 从本地缓存中查询
        String cacheValue = LOCAL_CACHE.getIfPresent(cacheKey);
        if (StrUtil.isNotBlank(cacheValue)){
            // 如果缓存命中
            return JSONUtil.toBean(cacheValue, Product.class);
        }

        // 4. 从 Redis 中查询
        cacheValue = stringRedisTemplate.opsForValue().get(redisKey);
        if (StrUtil.isNotBlank(cacheValue)){
            // 如果缓存命中, 先回写本地缓存
            Product product = JSONUtil.toBean(cacheValue, Product.class);
            LOCAL_CACHE.put(cacheKey, cacheValue);
            return product;
        }

        // 5. 查询数据库
        Product product = this.getById(id);
        if (product != null ){
            // 如果查询到结果，先回写 Redis，再回写 被本地缓存
            cacheValue = JSONUtil.toJsonStr(product);
            stringRedisTemplate.opsForValue().set(redisKey, cacheValue);
            LOCAL_CACHE.put(cacheKey, cacheValue);
            return product;
        }

        return product;
    }
}

结果测试

说明：这里为了展示结果，特意将 redis 、 mysql 数据库均部署在服务器上，以模拟实际环境，若部署在本地，会导致结果不明显

先插入一条数据在数据库中

先点击 controller 中的禁止小绿

得到

进行测试

第一次结果：497ms

第二次结果：18ms

第三次测试：5ms

后面继续测试效果不会有明显提升

结果分析

第一次查询时，本地缓存与 redis 中均没有缓存，会从 mysql 中查询，比较慢

第二次查询时，由于我点击较快，还没有将数据缓存在本地，从 redis 中查询，但速度获得大幅度提升

第三次查询时，缓存已经到本地，查询非常快

后续查询，缓存都在本地，查询速度没有明显提升

实战（高级）

基础代码写完后，在实际场景中会遇到 2 中的问题，下面我们分别会在代码中去解决，由于在查询时会用到MQ，因此，我们先讲数据一致性，再讲其它三个

数据一致性

延时双删

在 controller 中写入更新信息

▼
java
复制代码
/**
 * 修改商品信息
 * @param product 商品
 */
@PostMapping("/update")
public void updateProduct(@RequestBody Product product){
    productService.updateProduct(product);
}

在 Impl 中写入异步延时删除

▼
java
复制代码
// 注入异步线程池（避免使用主线程）
@Resource
private ThreadPoolTaskExecutor asyncTaskExecutor;

@Override
public void updateProduct(Product product) {
    // 1. 更新数据库
    this.updateById(product);


    // 2. 构建缓存Key（与查询时保持一致）
    Long productId = product.getId();
    String redisKey = DigestUtils.md5DigestAsHex(productId.toString().getBytes());
    String cacheKey = "cache" + redisKey;

    // 3. 删除 Caffeine
    LOCAL_CACHE.invalidate(cacheKey);
    System.out.println("本地缓存已删除：" + cacheKey);

    // 4. 删除Redis缓存
    Boolean delete = stringRedisTemplate.delete(redisKey);
    if (delete) {
        System.out.println("Redis缓存已删除：" + redisKey);
    }

    // 5. 异步延时双删除
    asyncTaskExecutor.execute(() -> {
        try {
            TimeUnit.SECONDS.sleep(3000);
            boolean delayDeleted = stringRedisTemplate.delete(redisKey);
            if (delayDeleted) {
                System.out.println("延时删除Redis缓存成功：" + redisKey);
            }
        }catch (Exception e) {
            System.err.println("延时删除发生异常：" + e.getMessage());
        }
    });

}

失败重试（使用 MQ）

引入依赖

▼
xml
复制代码
<!-- Rabbitmq -->
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

配置MQ

▼
java
复制代码
/**
 * 配置 RabbitMQ
 */
@Configuration
public class RabbitMQConfig {
    // 交换机名称（商品数据变更）
    public static final String PRODUCT_EXCHANGE = "product.exchange";
    // 队列名称（缓存失效通知）
    public static final String PRODUCT_CACHE_QUEUE = "product.cache.queue";
    // 路由键（匹配商品ID的变更通知）
    public static final String PRODUCT_CACHE_ROUTING_KEY = "product.cache.invalid.#";

    // 创建交换机
    @Bean
    public TopicExchange productExchange() {
        return new TopicExchange(PRODUCT_EXCHANGE, true, false);
    }

    // 创建队列（持久化）
    @Bean
    public Queue productCacheQueue() {
        return new Queue(PRODUCT_CACHE_QUEUE, true, false, false);
    }

    // 绑定交换机和队列
    @Bean
    public Binding bindingProductCacheQueue(TopicExchange productExchange, Queue productCacheQueue) {
        return BindingBuilder.bind(productCacheQueue).to(productExchange).with(PRODUCT_CACHE_ROUTING_KEY);
    }
}

编写消费者监听代码

▼
java
复制代码
@Component
public class ProductCacheInvalidListener {
    @Resource
    private StringRedisTemplate stringRedisTemplate;

    // 监听缓存失效队列
    @RabbitListener(queues = RabbitMQConfig.PRODUCT_CACHE_QUEUE)
    public void handleCacheInvalid(String redisKey) {
        boolean deleted = stringRedisTemplate.delete(redisKey);
        if (deleted) {
            System.out.println("Redis缓存删除成功：" + redisKey);
        } else {
            // Redis 删除失败，抛出异常触发 MQ 重试
            throw new RuntimeException("Redis缓存删除失败，触发MQ重试：" + redisKey);
        }
        // 如果均重试失败，还可以将其放入死信队列中，此处不再赘述
    }

}

编写生产者代码，在删除失败的时候将其放入消息队列中

▼
java
复制代码
// 引入 RabbitMQ
@Resource
private RabbitTemplate rabbitTemplate;

// 注入异步线程池（避免使用主线程）
@Resource
private ThreadPoolTaskExecutor asyncTaskExecutor;

@Override
public void updateProduct(Product product) {
    // 1. 更新数据库
    this.updateById(product);

    // 2. 构建缓存Key（与查询时保持一致）
    Long productId = product.getId();
    String redisKey = DigestUtils.md5DigestAsHex(productId.toString().getBytes());
    String cacheKey = "cache" + redisKey;

    // 3. 删除 Caffeine
    LOCAL_CACHE.invalidate(cacheKey);
    System.out.println("本地缓存已删除：" + cacheKey);

    // 4. 删除Redis缓存
    Boolean delete = stringRedisTemplate.delete(redisKey);
    if (delete) {
        System.out.println("Redis缓存已删除：" + redisKey);
    }

    // 5. 异步延时双删除
    asyncTaskExecutor.execute(() -> {
        try {
            TimeUnit.SECONDS.sleep(3000);
            boolean delayDeleted = stringRedisTemplate.delete(redisKey);
            if (delayDeleted) {
                System.out.println("延时删除Redis缓存成功：" + redisKey);
            }else {
                System.out.println("延时删除Redis缓存失败，准备放入MQ重试：" + redisKey);
                // 若延时删除仍失败，放入MQ重试（确保最终一致性）
                String routingKey = "product.cache.invalid.retry." + redisKey;
                rabbitTemplate.convertAndSend(
                        RabbitMQConfig.PRODUCT_EXCHANGE, // 交换机
                        routingKey, // 路由键
                        redisKey // 消息
                );
            }
        }catch (Exception e) {
            System.err.println("延时删除发生异常：" + e.getMessage());
        }
    });

}

订阅变更日志（使用 Canal）

引入依赖，此处使用1.1.8版本

▼
java
复制代码
<!--canal-->
<!-- Canal客户端核心依赖 -->
<dependency>
    <groupId>com.alibaba.otter</groupId>
    <artifactId>canal.client</artifactId>
    <version>1.1.8</version>
</dependency>

<!-- 显式引入protocol包（包含异常类定义） -->
<dependency>
    <groupId>com.alibaba.otter</groupId>
    <artifactId>canal.protocol</artifactId>
    <version>1.1.8</version>
</dependency>

编写配置文件

▼
java
复制代码
@Data
@Component
@ConfigurationProperties(prefix = "canal")  // 对应YML中的canal前缀
public class CanalProperties {

    private Server server = new Server();
    private String destination;
    private String username;
    private String password;
    private Listen listen = new Listen();

    // 内部类：服务端地址和端口
    @Data
    public static class Server {
        private String host;
        private int port;
    }

    // 内部类：监听的数据库和表
    @Data
    public static class Listen {
        private String database;
        private String table;
    }
}

▼
java
复制代码
@Configuration
public class CanalConfig {

    @Resource
    private CanalProperties canalProperties;

    @Bean
    public CanalConnector canalConnector() {
        // 从配置类中获取参数
        String host = canalProperties.getServer().getHost();
        int port = canalProperties.getServer().getPort();
        String destination = canalProperties.getDestination();
        String username = canalProperties.getUsername();
        String password = canalProperties.getPassword();

        // Canal 客户端的用户名/密码用于 Canal 服务端鉴权（通常默认不开启）。
        // 若未在 Canal 服务端开启鉴权，应当传 null，否则会因为鉴权失败而无法连接。

        // 创建Canal连接
        return CanalConnectors.newSingleConnector(
                new InetSocketAddress(host, port),
                destination,
                null,
                null
        );
    }
}

编写 Canal 监听器

▼
java
复制代码
@Component
public class CanalProductListener {

    // 本地缓存（Caffeine），假设已定义为全局常量
    public static final com.github.benmanes.caffeine.cache.Cache<String, Object> LOCAL_CACHE = 
            com.github.benmanes.caffeine.cache.Caffeine.newBuilder().build();

    @Resource
    private CanalConnector canalConnector;

    @Resource
    private StringRedisTemplate stringRedisTemplate;

    @Resource
    private CanalProperties canalProperties;

    @Resource
    private RabbitTemplate rabbitTemplate;

    // 线程池处理Canal事件（避免阻塞）
    private final ExecutorService executor = Executors.newSingleThreadExecutor();

    // 记录应用启动时间，用于过滤历史事件
    private final long appStartTimeMillis = System.currentTimeMillis();

    // 初始化时启动监听
    @PostConstruct
    public void startListener() {
        executor.submit(() -> {
            // 持续运行，异常后自动重连
            while (true) {
                try {
                    // 建立连接
                    canalConnector.connect();
                    System.out.println("Canal连接成功");

                    // 从配置中获取监听的数据库和表（动态生成订阅表达式）
                    String database = canalProperties.getListen().getDatabase();
                    String table = canalProperties.getListen().getTable();
                    String subscribeTable = database + "\\." + table;  // 如：your_db.product
                    canalConnector.subscribe(subscribeTable);  // 订阅指定表
                    canalConnector.rollback();

                    // 拉取并处理消息
                    while (true) {
                        Message message = canalConnector.getWithoutAck(100, 5000L, TimeUnit.MILLISECONDS);
                        long batchId = message.getId();
                        int size = message.getEntries().size();

                        if (batchId == -1 || size == 0) {
                            Thread.sleep(1000);
                            continue;
                        }

                        handleEntries(message.getEntries());
                        canalConnector.ack(batchId);
                    }

                } catch (Exception e) {
                    // 打印完整堆栈，便于定位 Read timed out 等问题
                    e.printStackTrace();
                    System.out.println("Canal连接失败或中断，5秒后重试...");
                    try {
                        TimeUnit.SECONDS.sleep(5);
                    } catch (InterruptedException ignored) {
                    }
                } finally {
                    try {
                        canalConnector.disconnect();
                    } catch (Exception ignored) {
                    }
                }
            }
        });
    }


    //处理Canal消息条目
    private void handleEntries(List<CanalEntry.Entry> entries) {
        for (CanalEntry.Entry entry : entries) {
            // 过滤非事务日志类型
            if (entry.getEntryType() != CanalEntry.EntryType.ROWDATA) {
                continue;
            }

            // 忽略应用启动前产生的历史事件，避免冷启动时清空缓存
            if (entry.getHeader() != null && entry.getHeader().getExecuteTime() > 0
                    && entry.getHeader().getExecuteTime() < appStartTimeMillis) {
                continue;
            }

            try {
                // 解析binlog日志
                CanalEntry.RowChange rowChange = CanalEntry.RowChange.parseFrom(entry.getStoreValue());
                CanalEntry.EventType eventType = rowChange.getEventType();

                // 只处理UPDATE事件（如果需要处理新增/删除，可添加EventType.INSERT/DELETE）
                if (eventType != CanalEntry.EventType.UPDATE) {
                    continue;
                }

                // 处理每行数据的变更
                for (CanalEntry.RowData rowData : rowChange.getRowDatasList()) {
                    handleProductUpdate(rowData.getAfterColumnsList()); // 取更新后的数据
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }


    // 处理商品更新事件，删除对应缓存
    private void handleProductUpdate(List<CanalEntry.Column> afterColumns) {
        // 从变更数据中获取productId（假设表的主键为id）
        Long productId = null;
        for (CanalEntry.Column column : afterColumns) {
            if ("id".equals(column.getName())) { // 匹配主键字段名
                productId = Long.parseLong(column.getValue());
                break;
            }
        }

        if (productId == null) {
            System.out.println("未找到productId，跳过缓存删除");
            return;
        }

        // 构建缓存Key（与原逻辑保持一致）
        String redisKey = DigestUtils.md5DigestAsHex(productId.toString().getBytes());
        String cacheKey = "cache" + redisKey;

        // 1. 删除本地缓存Caffeine
        LOCAL_CACHE.invalidate(cacheKey);
        System.out.println("Canal触发本地缓存删除：" + cacheKey);

        // 2. 删除Redis缓存
        Boolean delete = stringRedisTemplate.delete(redisKey);
        if (delete) {
            System.out.println("Canal触发Redis缓存删除：" + redisKey);
        } else {
            System.out.println("Redis缓存不存在或已删除：" + redisKey);
        }

        // 3. （可选）保留延时双删逻辑，进一步确保缓存一致性
        asyncDelayDelete(redisKey);
    }


    // 异步延时删除（与原逻辑一致，可选）
    private void asyncDelayDelete(String redisKey) {
        Executors.newSingleThreadExecutor().execute(() -> {
            try {
                TimeUnit.SECONDS.sleep(3); // 注意：原代码写的3000秒，这里修正为3秒（根据业务调整）
                boolean delayDeleted = stringRedisTemplate.delete(redisKey);
                if (delayDeleted) {
                    System.out.println("Canal延时删除Redis缓存成功：" + redisKey);
                } else {
                    System.out.println("Canal延时删除Redis缓存失败，放入MQ重试：" + redisKey);
                    // 放入MQ时携带时间戳头，供消费者过滤启动前的旧消息
                    String routingKey = "product.cache.invalid.retry." + redisKey;
                    rabbitTemplate.convertAndSend(
                            RabbitMQConfig.PRODUCT_EXCHANGE,
                            routingKey,
                            redisKey,
                            message -> {
                                message.getMessageProperties().setHeader("ts", System.currentTimeMillis());
                                return message;
                            }
                    );
                }
            } catch (Exception e) {
                System.err.println("Canal延时删除异常：" + e.getMessage());
            }
        });
    }
}

在编写业务代码时，仅仅需要更新数据库，无入侵代码

▼
java
复制代码
@Override
public void updateProductByCanal(Product product) {
    // 1. 更新数据库
    this.updateById(product);
}

分布式锁

由于分布式锁，不建议在此处使用，因此不过多赘述

过期兜底

任何一种方案，都可能会出现错误，于是我们应该在缓存上添加过期时间，且本地缓存 < Reids

本地缓存,当写入后，3分钟会过期

▼
java
复制代码
public final Cache<String, String> LOCAL_CACHE = Caffeine.newBuilder()
        .initialCapacity(1024) // 初始容量
        .maximumSize(10000L) // 最大缓存数量（超过后按 LRU 策略淘汰）
        .expireAfterWrite(3, TimeUnit.MINUTES) // 3 分钟后过期
        .recordStats() // 可选：记录缓存命中率等统计信息
        .build();

Redis 缓存，当存入缓存的时候可以进行设置

▼
java
复制代码
// 设置过期时间
int baseExpire = 5;
stringRedisTemplate.opsForValue().set(redisKey, cacheValue, baseExpire, TimeUnit.MINUTES);

缓存穿透

合法性检查

此处仅为演示，具体需要修改为业务所需，也可直接报异常

▼
java
复制代码
// 1. 合法性检查
if (id == null || id < 0){
    log.error("id:{}，非法",id);
    return null;
}

缓存空值

1. 先定义空值
2. 本地缓存中查询到空值需要判断并做相应的处理
3. Redis 中查询到空值需要判断，回写本地缓存，并做相应的处理
4. 数据库中查询不到数据时，需要将空值回写到 Redis 和本地缓存，并做相应的处理

▼
java
复制代码
private static final String NULL_PLACEHOLDER = "NULL_PLACEHOLDER";

@Override
public Product getProductDetail(Long id) {

    // 1. 合法性检查
    if (id == null || id < 0){
        log.error("id:{}，非法",id);
        return null;
    }

    // 2. 构建缓存key
    String redisKey = DigestUtils.md5DigestAsHex(id.toString().getBytes());
    String cacheKey = "cache" + redisKey;

    // 3. 从本地缓存中查询
    String cacheValue = LOCAL_CACHE.getIfPresent(cacheKey);
    if (StrUtil.isNotBlank(cacheValue)){
        // 处理空值标记
        if (NULL_PLACEHOLDER.equals(cacheValue)){
            log.warn("该值为空");
            return null;
        }
        // 如果缓存命中
        return JSONUtil.toBean(cacheValue, Product.class);
    }

    // 4. 从 Redis 中查询
    cacheValue = stringRedisTemplate.opsForValue().get(redisKey);
    if (StrUtil.isNotBlank(cacheValue)){
        if (NULL_PLACEHOLDER.equals(cacheValue)){
            // 回写本地缓存
            LOCAL_CACHE.put(cacheKey, cacheValue);
            log.warn("该值为空");
            return null;
        }
        // 如果缓存命中, 先回写本地缓存
        Product product = JSONUtil.toBean(cacheValue, Product.class);
        LOCAL_CACHE.put(cacheKey, cacheValue);
        return product;
    }

    // 5. 查询数据库
    Product product = this.getById(id);
    if (product != null ){
        // 如果查询到结果，先回写 Redis，再回写 被本地缓存
        cacheValue = JSONUtil.toJsonStr(product);

        // 设置过期时间
        int baseExpire = 5;
        stringRedisTemplate.opsForValue().set(redisKey, cacheValue, baseExpire, TimeUnit.MINUTES);

        LOCAL_CACHE.put(cacheKey, cacheValue);
        return product;
    }else {
        // 6. 如果查询数据不存在，缓存空值
        int baseExpire = 5;
        stringRedisTemplate.opsForValue().set(redisKey, NULL_PLACEHOLDER,baseExpire, TimeUnit.MINUTES);
        LOCAL_CACHE.put(cacheKey, NULL_PLACEHOLDER);
        return null;
    }
}

布隆过滤器

todo 由于该插件版本问题，暂未实践，但上述原理已经讲清楚，也可自己实现一个过滤器，待续.....

缓存击穿

分布式锁

引入依赖

▼
java
复制代码
<!-- Redisson -->
<dependency>
    <groupId>org.redisson</groupId>
    <artifactId>redisson</artifactId>
    <version>3.50.0</version>
</dependency>

添加配置

▼
java
复制代码
@Configuration
public class RedissonConfig {

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

    @Value("${spring.data.redis.port}")
    private Integer redisPort;

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

    @Value("${spring.data.redis.database}")
    private Integer redisDatabase;

    @Bean
    public RedissonClient redissonClient() {
        Config config = new Config();
        String address = "redis://" + redisHost + ":" + redisPort;
        SingleServerConfig singleServerConfig = config.useSingleServer()
        .setAddress(address)
        .setDatabase(redisDatabase)
        .setConnectionMinimumIdleSize(1)
        .setConnectionPoolSize(10)
        .setIdleConnectionTimeout(30000)
        .setConnectTimeout(5000)
        .setTimeout(3000)
        .setRetryAttempts(3)
        .setRetryInterval(1500);
        // 如果有密码则设置密码
        if (redisPassword != null && !redisPassword.isEmpty()) {
            singleServerConfig.setPassword(redisPassword);
        }
        return Redisson.create(config);
    }
}

编写代码

▼
java
复制代码
/**
 * 通过分布式锁查询数据库
 * @param id 查询条件
 */
private Product getProductByRedisson(Long id, String redisKey, String cacheKey) {
    String cacheValue;
    Product product = null;
    // 定义锁的 key（建议加上业务前缀，避免冲突）
    String lockKey = "lock:product:detail:" + id;
    // 获取锁对象
    RLock lock = redissonClient.getLock(lockKey);

    try {
        // 尝试获取锁：最多等待 100ms，10秒后自动释放（防止死锁）
        // 注意：Redisson 的看门狗机制会自动续期，只要线程未释放锁且未宕机，锁不会过期
        boolean isLocked = lock.tryLock(100, 10, TimeUnit.SECONDS);
        if (isLocked) {
            // 成功获取锁后，再次检查 Redis 缓存（避免其他线程已重建缓存）
            String doubleCheckValue = stringRedisTemplate.opsForValue().get(redisKey);
            if (StrUtil.isNotBlank(doubleCheckValue)) {
                return handleCacheHit(cacheKey, doubleCheckValue);
            }

            // 真正查询数据库
            product = this.getById(id);

            // 7. 回写缓存（设置随机过期时间，避免缓存雪崩）
            if (product != null) {
                cacheValue = JSONUtil.toJsonStr(product);
            } else {
                // 缓存空值，避免缓存穿透
                cacheValue = NULL_PLACEHOLDER;
            }

            // 过期时间设置为 5 分钟
            int randomExpire = 5;
            stringRedisTemplate.opsForValue().set(redisKey, cacheValue, randomExpire, TimeUnit.MINUTES);
            // 回写本地缓存
            LOCAL_CACHE.put(cacheKey, cacheValue);
        } else {
            // 未获取到锁，休眠 50-100ms 后重试（避免频繁重试）
            Thread.sleep(50 + new Random().nextInt(51));
            return getProductDetailByRedisson(id);
        }
    } catch (InterruptedException e) {
        log.error("获取锁或重试失败", e);
        Thread.currentThread().interrupt(); // 恢复中断状态
        return null;
    } finally {
        // 释放锁（只有持有锁的线程才能释放）
        if (lock.isHeldByCurrentThread()) {
            lock.unlock();
        }
    }

    return product;
}


// 封装缓存命中处理逻辑（简化代码）
private Product handleCacheHit(String cacheKey, String cacheValue) {
    if (NULL_PLACEHOLDER.equals(cacheValue)) {
        LOCAL_CACHE.put(cacheKey, cacheValue);
        log.warn("该值为空");
        return null;
    }
    Product product = JSONUtil.toBean(cacheValue, Product.class);
    LOCAL_CACHE.put(cacheKey, cacheValue);
    return product;
}

永不过期 + 主动更新

不设置过期时间即可，主动更新上面mq已经写过了

逻辑过期 + 被动更新

定义带逻辑时间的缓存结构

▼
java
复制代码
// 定义带逻辑过期时间的缓存结构
@Data
private static class LogicExpireCache {
    private Product data; // 实际业务数据
    private long expireTime; // 逻辑过期时间（毫秒时间戳）
}

修改缓存命中的处理方法

▼
java
复制代码
// 处理缓存命中逻辑（核心：判断逻辑过期并触发异步更新）
private Product handleLogicCacheHit(String cacheKey, String cacheValue) {
    // 空值处理（保持不变）
    if (NULL_PLACEHOLDER.equals(cacheValue)) {
        LOCAL_CACHE.put(cacheKey, cacheValue);
        log.warn("该值为空");
        return null;
    }

    // 解析带逻辑过期时间的缓存数据
    LogicExpireCache cacheData = JSONUtil.toBean(cacheValue, LogicExpireCache.class);
    Product product = cacheData.getData();
    long expireTime = cacheData.getExpireTime();

    // 判断是否逻辑过期
    if (System.currentTimeMillis() < expireTime) {
        // 未过期：直接返回数据，回写本地缓存
        LOCAL_CACHE.put(cacheKey, cacheValue);
        return product;
    } else {
        // 已过期：返回旧数据，同时异步更新缓存（不阻塞当前请求）
        LOCAL_CACHE.put(cacheKey, cacheValue); // 本地缓存先保留旧数据
        asyncUpdateLogicCache(cacheKey, cacheData.getData().getId()); // 异步更新
        return product; // 先返回旧数据，保证响应速度
    }
}

编写核心流程

▼
java
复制代码
@Override
public Product getProductDetailByLogicExpire(Long id) {

    // 1. 合法性检查
    if (id == null || id < 0) {
        log.error("id:{}，非法", id);
        return null;
    }

    // 2. 构建缓存key
    String redisKey = DigestUtils.md5DigestAsHex(id.toString().getBytes());
    String cacheKey = "cache" + redisKey;

    // 3. 从本地缓存中查询（核心：处理逻辑过期）
    String cacheValue = LOCAL_CACHE.getIfPresent(cacheKey);
    if (StrUtil.isNotBlank(cacheValue)) {
        return handleLogicCacheHit(cacheKey, cacheValue);
    }

    // 4. 从 Redis 中查询（核心：处理逻辑过期）
    cacheValue = stringRedisTemplate.opsForValue().get(redisKey);
    if (StrUtil.isNotBlank(cacheValue)) {
        return handleLogicCacheHit(cacheKey, cacheValue);
    }

    // 5. Redis缓存未命中（首次查询或缓存被意外删除）
    // 直接查库并初始化逻辑过期缓存（无需加锁，首次查询压力低）
    Product product = this.getById(id);
    if (product != null) {
        // 封装带逻辑过期时间的缓存数据（设置5分钟后逻辑过期）
        LogicExpireCache cacheData = new LogicExpireCache();
        cacheData.setData(product);
        cacheData.setExpireTime(System.currentTimeMillis() + 5 * 60 * 1000); // 5分钟后过期
        String jsonValue = JSONUtil.toJsonStr(cacheData);
        // Redis存储时不设置物理过期（永不过期）
        stringRedisTemplate.opsForValue().set(redisKey, jsonValue);
        // 回写本地缓存
        LOCAL_CACHE.put(cacheKey, jsonValue);
        return product;
    } else {
        // 缓存空值（避免缓存穿透，设置物理过期，防止长期占用空间）
        stringRedisTemplate.opsForValue().set(redisKey, NULL_PLACEHOLDER, 5, TimeUnit.MINUTES);
        LOCAL_CACHE.put(cacheKey, NULL_PLACEHOLDER);
        return null;
    }
}

// 异步更新逻辑过期缓存（加简单锁避免并发更新）
@Async // 需要开启@EnableAsync注解
public void asyncUpdateLogicCache(String cacheKey, Long id) {
    String redisKey = DigestUtils.md5DigestAsHex(id.toString().getBytes());
    String lockKey = "lock:product:update:" + id; // 异步更新的锁

    RLock lock = redissonClient.getLock(lockKey);
    try {
        // 尝试获取锁，最多等1秒，持有3秒（防止更新逻辑卡住）
        boolean isLocked = lock.tryLock(1, 3, TimeUnit.SECONDS);
        if (isLocked) {
            // 查询最新数据
            Product newProduct = this.getById(id);
            if (newProduct != null) {
                // 生成新的逻辑过期时间（续5分钟）
                LogicExpireCache newCacheData = new LogicExpireCache();
                newCacheData.setData(newProduct);
                newCacheData.setExpireTime(System.currentTimeMillis() + 5 * 60 * 1000);
                String newJsonValue = JSONUtil.toJsonStr(newCacheData);
                // 更新Redis和本地缓存
                stringRedisTemplate.opsForValue().set(redisKey, newJsonValue);
                LOCAL_CACHE.put(cacheKey, newJsonValue);
            } else {
                // 数据已删除，缓存空值（物理过期）
                stringRedisTemplate.opsForValue().set(redisKey, NULL_PLACEHOLDER, 5, TimeUnit.MINUTES);
                LOCAL_CACHE.put(cacheKey, NULL_PLACEHOLDER);
            }
        }
    } catch (InterruptedException e) {
        log.error("异步更新缓存失败", e);
        Thread.currentThread().interrupt();
    } finally {
        if (lock.isHeldByCurrentThread()) {
            lock.unlock();
        }
    }
}

缓存雪崩

过期时间加随机值，多级缓存时间错开

由于 Caffeine 没有提供可随意添加过期时间的 put 方法，我自定义了一个工具类，用于添加随机时间

▼
java
复制代码
/**
 * Caffeine 缓存工具类，支持动态设置每个 key 的过期时间
 */
public class CaffeineCacheUtil<K, V> {

    // 底层 Caffeine 缓存
    private final Cache<K, V> cache;

    // 存储每个 key 的过期时间（纳秒），用于动态调整
    private final Map<K, Long> keyExpireMap = new ConcurrentHashMap<>();

    // 随机数生成器（复用避免性能损耗）
    private final Random random = new Random();

    /**
     * 初始化缓存
     * @param initialCapacity 初始容量
     * @param maximumSize 最大容量
     */
    public CaffeineCacheUtil(int initialCapacity, long maximumSize) {
        // 构建缓存，使用自定义 Expiry 策略
        this.cache = Caffeine.newBuilder()
                .initialCapacity(initialCapacity)
                .maximumSize(maximumSize)
                .expireAfter(new CustomExpiry<>()) // 自定义过期策略
                .recordStats()
                .build();
    }

    /**
     * 添加缓存（永不过期）
     */
    public void put(K key, V value) {
        // 不设置过期时间，不在 keyExpireMap 中存储，视为永不过期
        cache.put(key, value);
    }

    /**
     * 添加缓存（使用默认过期时间：3分钟 + 0-300秒随机）
     */
    public void putWithDefaultExpire(K key, V value) {
        // 生成默认随机过期时间（3-8分钟）
        long baseExpire = TimeUnit.MINUTES.toNanos(3); // 3分钟（纳秒）
        long randomOffset = TimeUnit.SECONDS.toNanos(random.nextInt(301)); // 0-300秒随机
        long expireNanos = baseExpire + randomOffset;
        put(key, value, expireNanos, TimeUnit.NANOSECONDS);
    }

    /**
     * 添加缓存（自定义过期时间）
     */
    public void put(K key, V value, long duration, TimeUnit unit) {
        // 转换为纳秒并存储
        long expireNanos = unit.toNanos(duration);
        keyExpireMap.put(key, expireNanos);
        // 放入缓存
        cache.put(key, value);
    }

    /**
     * 获取缓存
     */
    @Nullable
    public V getIfPresent(K key) {
        return cache.getIfPresent(key);
    }

    /**
     * 移除缓存
     */
    public void invalidate(K key) {
        cache.invalidate(key);
        keyExpireMap.remove(key);
    }

    /**
     * 清空缓存
     */
    public void invalidateAll() {
        cache.invalidateAll();
        keyExpireMap.clear();
    }

    /**
     * 获取缓存统计信息
     */
    public String stats() {
        return cache.stats().toString();
    }

    /**
     * 自定义过期策略：从 keyExpireMap 中获取每个 key 的过期时间
     */
    private class CustomExpiry<K, V> implements Expiry<K, V> {

        @Override
        public long expireAfterCreate(@NonNull K key, @NonNull V value, long currentTime) {
            // 创建时，从 map 中获取预设的过期时间（默认 0 表示立即过期，避免未设置的 key 永不过期）
            return keyExpireMap.getOrDefault(key, 0L);
        }

        @Override
        public long expireAfterUpdate(@NonNull K key, @NonNull V value, long currentTime, @NonNegative long currentDuration) {
            // 更新时，使用新设置的过期时间（若未重新设置则沿用旧值）
            return keyExpireMap.getOrDefault(key, currentDuration);
        }

        @Override
        public long expireAfterRead(@NonNull K key, @NonNull V value, long currentTime, @NonNegative long currentDuration) {
            // 读取时不改变过期时间（可根据需求改为延长过期）
            return currentDuration;
        }
    }
}

封装生成随机过期时间的方法

▼
java
复制代码
/**
 * 生成随机过期时间（秒）
 * @param maxRandomSeconds 随机数的最大范围（秒），必须为非负数
 * @return 0 到 maxRandomSeconds（包含）之间的随机秒数
 */
public static int generateRandomExpireSeconds(int maxRandomSeconds) {
    // 校验参数：若传入负数，默认返回0（避免异常）
    if (maxRandomSeconds < 0) {
        return 0;
    }
    // 生成 0 到 maxRandomSeconds（包含）的随机整数
    return RANDOM.nextInt(maxRandomSeconds + 1);
}

修改创建 Caffeine 的代码

▼
java
复制代码
// 使用自定义本地缓存工具类创建本地缓存
public final CaffeineCacheUtil<String, String> LOCAL_CACHE = new CaffeineCacheUtil<>(1024, 10000L);

在添加 Caffeine 缓存的代码中加入随机过期时间

▼
java
复制代码
LOCAL_CACHE.put(cacheKey, cacheValue, 180 + generateRandomExpireSeconds(180),TimeUnit.SECONDS);

缓存集群化部署

todo 文字描述繁琐，待补

总结

针对查询

其实使用查询的流程可以是一个模板方法，现在我们用图的方式来一步步深刻理解其流程。

分布式锁策略

特性：

缓存穿透：合法性检查、缓存空值

缓存击穿：分布式锁

缓存雪崩：过期时间随机化，Caffeine < Redis

逻辑过期策略

特性：

缓存穿透：合法性检查、缓存空值

缓存击穿：逻辑过期

缓存雪崩：需要通过其它方式（集群部署、降级熔断）

异步更新缓存，这个操作是异步的，且需要上分布式锁，但是这个影响没有前面那么大：

针对修改

在第 2 章节，已经对其做了详细的讲解，因此，本节做最后的总结

MQ

在更新数据时使用延时双删，删除失败后，生产消息给 MQ 消费者，处理消息，就进行删除缓存

Canal

需要进行配置Canal才行