1、概述

在使用Redis作为MySQL的缓存层时，缓存一致性问题是指Redis中的缓存数据与MySQL数据库中的实际数据不一致的情况。这可能会导致读取到过期或错误的数据，从而影响系统的正确性和用户体验。
为了减轻数据库的压力，通常读操作都是先读缓存，缓存没有则读数据库数据在写入缓存；而增/删/改操作介于数据库和缓存之间，由于操作步骤和并发问题，可能产生不一致的现象。

2、缓存一致性问题的表现

• 脏读：客户端从Redis中读取到的是旧数据或过期数据，而MySQL中的数据已经发生了变化。

3、缓存一致性问题的原因

- 缓存更新不及时：当MySQL中的数据发生变化时，Redis中的缓存没有及时更新或删除，导致客户端读取到过期数据。
- 缓存失效策略不合理：如果缓存的TTL（生存时间）设置不当，可能会导致缓存过早或过晚失效，进而引发一致性问题。
- 并发写入冲突：在高并发场景下，多个客户端同时对同一数据进行写操作，可能导致缓存和数据库之间的数据不一致。

4、解决缓存一致性问题的方法

为了确保Redis和MySQL之间的数据一致性，可以采用以下几种常见的解决方案：

（1）、更新数据库时同步更新缓存（Write Through）

- 原理：

• 在更新MySQL数据的同时，立即更新Redis中的缓存。这样可以确保缓存中的数据始终与数据库保持一致。

- 优点：

• 简单易实现，能够保证强一致性。

- 缺点：

• 写操作的性能会受到影响，因为每次写操作都需要同时更新数据库和缓存。
• 高并发下Redis写操作结果的不确定性，很可能造成非预期的结果。（删除却能保证结果一致）
• Redis的写操作可能会造成底层数据结构的改变，造成额外时间开销。如（List的压缩列表转双向列表）。

- 适用场景：

• 适用于对数据一致性要求较高的场景，尤其是写操作较少的系统。

（2）、更新数据库后删除缓存（Write Behind）（推荐）

- 原理：

• 在更新MySQL数据后，立即将Redis中对应的缓存键删除。下次读取时，Redis会发现缓存已失效，重新从MySQL中加载最新的数据并更新缓存。

- 优点：

• 写操作的性能较高，因为只需要更新数据库，不需要立即更新缓存。
• 避免了缓存不一致问题（高并发场景下更新缓存可能造成缓存结果不确定，但是删除操作结果是确定的）。

- 缺点：

• 存在短暂的时间窗口，期间可能会读取到旧数据（弱一致性）。
• 可能会触发缓存击穿，尤其是在高并发场景下。

- 适用场景：

• 适用于对数据一致性要求不高，但对写性能要求较高的场景。

代码示例：

import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import javax.persistence.EntityManager;
import java.util.concurrent.TimeUnit;

@Service
public class ProductService {

    @Autowired
private ProductRepository productRepository; 

    @Autowired
private RedisTemplate<String, Object> redisTemplate;  

    @Autowired
private RedissonClient redissonClient;  

    @Autowired
    private EntityManager entityManager;  // 数据库

    // Redis 锁前缀
    private static final String LOCK_PREFIX = "product:lock:";

    // 缓存键前缀
    private static final String CACHE_KEY_PREFIX = "product:cache:";

    /**
     * 更新产品信息，并确保缓存一致性
     * @param productId 产品ID
     * @param newPrice  新的价格
     */
    @Transactional
    public void updateProductPrice(Long productId, double newPrice) {
        // 1. 获取分布式锁，确保同一时间只有一个线程可以更新该产品的价格
        RLock lock = redissonClient.getLock(LOCK_PREFIX + productId);
        try {
            // 尝试获取锁，最多等待5秒，锁的持有时间为10秒
            if (lock.tryLock(5, 10, TimeUnit.SECONDS)) {
                // 2. 开始数据库事务，更新产品价格
                Product product = productRepository.findById(productId)
                        .orElseThrow(() -> new RuntimeException("Product not found"));

                // 更新产品价格
                product.setPrice(newPrice);
                productRepository.save(product);

                // 3. 删除Redis中的缓存，确保下次读取时能够从数据库中获取最新的数据
                redisTemplate.delete(CACHE_KEY_PREFIX + productId);

                // 4. 手动刷新实体管理器，确保事务提交后的数据一致性
                entityManager.flush();
            } else {
                throw new RuntimeException("Failed to acquire lock for product " + productId);
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            throw new RuntimeException("Interrupted while trying to acquire lock", e);
        } finally {
            // 5. 释放锁
            if (lock.isHeldByCurrentThread()) {
                lock.unlock();
            }
        }
    }

    /**
     * 获取产品信息，优先从缓存中读取，如果缓存不存在则从数据库中读取并更新缓存
     *
     * @param productId 产品ID
     * @return 产品信息
     */
    public Product getProductById(Long productId) {
        // 1. 尝试从 Redis 缓存中获取产品信息
        String cacheKey = CACHE_KEY_PREFIX + productId;
        Product cachedProduct = (Product) redisTemplate.opsForValue().get(cacheKey);

        if (cachedProduct != null) {
            // 2. 如果缓存存在，直接返回缓存中的数据
            return cachedProduct;
        }

        // 3. 如果缓存不存在，从数据库中获取产品信息
        Product product = productRepository.findById(productId)
                .orElseThrow(() -> new RuntimeException("Product not found"));

        // 4. 使用分布式锁，确保只有一个线程能够更新缓存
        RLock lock = redissonClient.getLock(LOCK_PREFIX + productId);
        try {
            // 尝试获取锁，最多等待5秒，锁的持有时间为10秒
            if (lock.tryLock(5, 10, TimeUnit.SECONDS)) {
                // 5. 再次检查缓存，防止其他线程已经更新了缓存
                cachedProduct = (Product) redisTemplate.opsForValue().get(cacheKey);
                if (cachedProduct == null) {
                    // 6. 如果缓存仍然不存在，将数据库中的数据写入缓存
                    redisTemplate.opsForValue().set(cacheKey, product, 60, TimeUnit.MINUTES);  // 设置缓存过期时间为60分钟
                }
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            throw new RuntimeException("Interrupted while trying to acquire lock", e);
        } finally {
            // 7. 释放锁
            if (lock.isHeldByCurrentThread()) {
                lock.unlock();
            }
        }

        // 8. 返回产品信息
        return product;
    }
}

（3）、为什么先更新数据库，后更新缓存呢？

1、先更新数据库，再更新缓存（Write Behind）

**- 原理：**在写操作时，首先更新MySQL数据库中的数据，然后更新Redis缓存中的数据。这样可以确保数据库中的数据是最新的，即使缓存更新失败，数据库中的数据仍然是正确的。

- 优点：

• 数据安全：数据库中的数据始终是最新的，确保最终数据的正确性和安全性。
• 容错性好：如果Redis更新失败或Redis服务不可用，系统仍然可以依赖MySQL中的数据，不会导致数据丢失。
• 简化回滚逻辑：如果写操作失败，只需回滚数据库中的事务，而不需要同时处理缓存的回滚，降低了复杂性。

- 缺点：

• 短暂的不一致：在数据库更新成功但缓存尚未更新的时间窗口内，客户端可能会读取到旧的缓存数据。这个时间窗口的长度取决于缓存更新的延迟（通常比较短，可以接受）。
• 并发写入冲突：在高并发场景下，多个客户端可能同时对同一数据进行写操作，导致缓存更新的竞争问题。可以通过分布式锁等机制解决，但会增加系统复杂度。
• 写放大问题：每次写操作都需要同时更新数据库和缓存，增加了写操作的开销，尤其是在高并发场景下，可能会对性能产生一定影响。

- 适用场景：

• 对数据一致性要求较高：如果你的应用对数据一致性要求较高，尤其是不允许读取到过期数据，那么先更新数据库再更新缓存是更好的选择。
• 容错性要求高：如果你希望即使 Redis 出现故障，系统仍然能够正常运行并依赖数据库中的最新数据，那么这种方案更合适。

2、先更新缓存，再更新数据库（Write Through）

- 原理：在写操作时，首先更新Redis缓存中的数据，然后再更新MySQL数据库中的数据。这样可以确保客户端在写操作完成后立即读取到最新的数据，避免了短暂的不一致问题。

- 优点：

• 避免短暂不一致：客户端在写操作完成后立即可以读取到最新的数据，避免了短暂的不一致问题。
• 减少缓存击穿：由于缓存已经提前更新，后续的读请求可以直接从Redis中获取最新的数据，减少了缓存击穿的可能性。

- 缺点：

• 数据丢失风险：如果Redis更新成功但MySQL更新失败，可能会导致数据丢失或不一致。此时，Redis中的数据是最新的，但MySQL中的数据仍然是旧的。
• 复杂的回滚逻辑：如果写操作失败，需要同时回滚Redis和MySQL中的数据，增加了系统的复杂性。特别是当Redis和MySQL之间的事务无法原子化时，可能会导致部分更新成功、部分更新失败的情况。
• 缓存污染：如果Redis更新成功但MySQL更新失败，Redis中的缓存可能会被污染，导致后续读取到错误的数据。为了解决这个问题，通常需要引入额外的机制（如消息队列、分布式锁等）来确保缓存和数据库的一致性。

- 适用场景：

• 读操作占主导：如果你的应用以读操作为主，写操作较少，先更新缓存可以确保读操作的性能和一致性。
• 容忍一定的数据丢失风险：如果你的应用可以容忍一定的数据丢失风险，或者有其他机制（如定期同步、备份等）来确保数据的最终一致性，那么这种方案是可以考虑的。

3、最佳实践：结合两者的优势

*先更新数据库，再删除缓存通常是最优的方法，也是最常用的做法。*写操作时，首先更新MySQL数据库中的数据，然后删除Redis中对应的缓存键。下次读取时，Redis会发现缓存已失效，重新从MySQL中加载最新的数据并更新缓存。这种方法既保证了数据库中的数据始终是最新的，又避免了缓存和数据库不一致的问题。

- 优点：

• 强一致性：数据库中的数据始终是最新的，避免了数据丢失的风险。
• 简化回滚逻辑：如果写操作失败，只需回滚数据库中的事务，而不需要同时处理缓存的回滚。
• 减少缓存污染：即使Redis更新失败，也不会导致缓存污染，因为缓存已经被删除。

- 缺点：

• 短暂的不一致：在数据库更新成功但缓存尚未更新的时间窗口内，客户端可能会读取到旧的缓存数据。（但这个时间通常很短可以接受）
• 缓存击穿风险：如果大量并发请求同时访问同一个缓存键，可能会导致缓存击穿。但可以通过引入缓存预热、分布式锁等机制来缓解这个问题。

在绝大部分的系统中，数据安全永远才是第一位的，如果以牺牲数据安全为代价来提升系统性能通常都是不可取的。为了保障数据的安全，一般都要将数据保存到数据库中，而不是保存在缓存中（丢失风险大）。缓存最根本的目的是为了提升系统的查询的效率，减轻数据库的查询负担。如果成功更新了缓存，但是在执行更新数据库时服务器突然宕机了。此时缓存中是最新数据，数据库中仍然是旧数据，从数据安全的角度来说就是丢失了数据。所以通常建议一定是先更新数据库，保证数据安全不丢失为第一位。

（4）、其他优化方案

通常我们使用先更新数据库后删除缓存（如上4.2）的方式就足够了。此外还有一些其他优化的方式可以了解下。

1、消息队列MQ

对于一些分布式的场景，可以使用消息队列来解耦MySQL和Redis的写入操作。

在同时操作缓存和数据库时，都无法保证两者都能一次性操作成功，所以我们最好的办法就是重试，这个重试并不是立即重试，因为缓存和数据库可能因为网络或者其它原因停止服务了，立即重试成功率极低，而且重试会占用线程资源，显然不合理，所以我们需要采用异步重试机制。

异步重试我们可以使用消息队列来完成，因为消息队列可以保证消息的可靠性，消息不会丢失，也可以保证正确消费，当且仅当消息消费成功后才会将消息从消息队列中删除。

说明下：
这种方式需要介入MQ（如RocketMQ、Kafka 2.5+），虽然发布消息到消息队列的速度比直接删除Redis键的速度要慢。但是消息队列可以保证消息的可靠性，提供了异步重试机制，保证任务执行成功后才会删除任务。如果我们把删除Redis键的任务交给消息队列就可以确保成功，避免了Redis直接删除键失败的情况。

个人觉得：这种方式安全性比较好，但实现消息队列带来的成本比较大，也更复杂。仅用消息队列去删除Redis键，实际比直接删除更慢，而且Redis删除key失败的情况非常低，通常没有必要这么做。

2、Canal+Binlog同步

Canal是一个基于MySQL Binlog的增量数据同步工具。它通过监听MySQL的Binlog日志，捕获所有的数据变更（如插入、更新、删除）。当数据库发生变更时，canal就可以帮我们拿到具体操作的数据，然后再去根据具体的数据，去删除对应的缓存。
通过这种方式，我们仅需要关注mysql的修改，无需关心缓存的修改。当修改一条mysql的数据时，mysql就会生成一条binlog日志，我们可以通过Canal订阅这种消息，拿到具体修改的数据，之后就可以在更新缓存了。订阅日志目前比较流行的就是阿里开源的Canal。
注意：Canal本身是没有数据处理能力的，我们可以结合Canal +消息队列一起来使用，从而达到实现更新缓存的操作。
原理示意图：

优点：

• 自动同步：无需手动编写代码来同步数据，Canal会自动捕获MySQL的变更并同步到Redis。
• 低延迟：Canal可以实时捕获MySQL的变更，确保Redis和MySQL之间的数据同步延迟较低。
• 最终一致性：虽然不能保证强一致性，但可以通过Canal的重试机制和幂等性设计来保证最终一致性。

缺点：

• 依赖MySQL的Binlog：Canal需要MySQL开启 Binlog，并且必须使用ROW格式的Binlog，否则无法捕获详细的变更信息。
• 单点故障：Canal本身可能存在单点故障，建议使用Canal的集群模式或多实例部署来提高可用性。

标题扩展介绍下Canal：

1、概念

Canal是阿里巴巴开源的一款基于MySQL数据库增量日志解析的工具，它能够实时捕获MySQL的Binlog（二进制日志），并将这些变更事件转发到其他系统（如Kafka、Redis、Elasticsearch等）。
Canal的核心功能是通过模拟MySQL主从复制协议，监听MySQL的Binlog日志，从而实现数据的实时同步。

2、Canal监听MySQL日志的原理

（1）、模拟MySQL主从复制：

• Canal通过MySQL的主从复制协议与MySQL建立连接。它模拟了一个MySQL从库的行为，向MySQL发送SHOW MASTER STATUS和SHOW SLAVE STATUS等命令，获取当前的Binlog文件名和位置。

（2）、订阅Binlog事件：

• Canal使用MySQL提供的binlog dump协议，订阅MySQL的Binlog事件。MySQL会将所有的DDL（数据定义语言）和DML（数据操作语言）操作（如INSERT、UPDATE、DELETE）以二进制日志的形式发送给Canal。

（3）、解析Binlog事件：

• Canal接收到Binlog事件后，会解析这些二进制日志，提取出具体的表结构变化和数据变更信息。Canal支持多种解析格式，包括Row-based、Statement-based和Mixed-based。

（4）、转发变更事件：

• 解析后的变更事件可以通过Canal的插件机制，转发到其他系统（如Kafka、Redis、Elasticsearch等），或者直接在应用程序中处理。

3、Canal的架构

Canal 的架构主要包括以下几个组件：

• Canal Server：负责与MySQL建立连接，监听Binlog日志，并将解析后的变更事件转发给下游系统。
• Canal Client：负责接收Canal Server发送的变更事件，并进行相应的处理。
• Canal Adapter：用于将Canal解析的变更事件转发到不同的目标系统（如Kafka、Redis、Elasticsearch等）。

个人觉得：这个方法，首先需要mysql启用binlog日志。还需要我们下载和安装Canal，在配置并启动Canal。然后代码端还要集成Canal的实现。可谓是既费时又费劲，如果只是为了实现删除缓存，个人感觉真的没有必要。