当前位置： 首页 > article >正文

JavaEE企业级开发 延迟双删+版本号机制(乐观锁) 事务保证redis和mysql的数据一致性 示例

article 2025/3/26 11:30:34

提醒

要求了解或者熟练掌握以下知识点

  1. spring 事务
  2. mysql 脏读
  3. 如何保证缓存和数据库数据一致性
  4. 延迟双删
  5. 分布式锁
  6. 并发编程 原子操作类

前言

在起草这篇博客之前

我做了点功课

这边我写的是一个示例代码

数据层都写成了 mock 的形式(来源于 JUnit5)

// Dduo
import java.util.concurrent.ConcurrentHashMap;  
import java.util.concurrent.Executors;  
import java.util.concurrent.ScheduledExecutorService;  
import java.util.concurrent.TimeUnit;  
 
// 数据服务类 
public class DataService { 

    // 模拟缓存（实际使用Redis等实现） 
    private static final ConcurrentHashMap<String, String> cache = new ConcurrentHashMap<>(); 
    // 延迟双删线程池 
    private final ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1);  
    // 模拟数据库，使用一个 Map 来存储数据记录 
    private static final ConcurrentHashMap<Integer, DataRecord> mockDatabase = new ConcurrentHashMap<>(); 
 
    // 数据记录类，包含数据的基本信息和版本号 
    private static class DataRecord { 
        private int id; 
        private String content; 
        private int version; 
 
        public DataRecord(int id, String content, int version) { 
            this.id  = id; 
            this.content  = content; 
            this.version  = version; 
        } 
 
        public int getId() { 
            return id; 
        } 
 
        public String getContent() { 
            return content; 
        } 
 
        public void setContent(String content) { 
            this.content  = content; 
        } 
 
        public int getVersion() { 
            return version; 
        } 
 
        public void setVersion(int version) { 
            this.version  = version; 
        } 
    } 
 
    // 模拟从数据库获取数据 
    private static DataRecord mockDatabaseGet(int id) { 
        return mockDatabase.get(id);  
    } 
 
    // 模拟数据库更新操作，更新数据并更新版本号 
    private static boolean mockDatabaseUpdate(int id, String content, int expectedVersion) { 
        DataRecord record = mockDatabase.get(id);  
        if (record == null) { 
            return false; 
        } 
        // 检查版本号是否匹配 
        if (record.getVersion()  != expectedVersion) { 
            return false; 
        } 
        // 更新数据内容 
        record.setContent(content);  
        // 更新版本号 
        record.setVersion(expectedVersion  + 1); 
        mockDatabase.put(id,  record); 
        return true; 
    } 
 
    // 初始化数据库数据 
    public void initData(int id, String content) { 
        mockDatabase.put(id,  new DataRecord(id, content, 1)); 
    } 
 
    // 获取数据（带缓存逻辑） 
    public String getData(int id) { 
        String cacheKey = "data_" + id; 
        // 1. 先查缓存 
        String cached = cache.get(cacheKey);  
        if (cached != null) { 
            return cached; 
        } 
 
        // 2. 缓存未命中，查询数据库 
        DataRecord record = mockDatabaseGet(id); 
        if (record == null) { 
            return null; 
        } 
 
        // 3. 写入缓存（包含版本号信息） 
        String value = record.getContent()  + "|v" + record.getVersion();  
        cache.put(cacheKey,  value); 
        return value; 
    } 
 
    // 更新数据（带延迟双删和版本控制） 
    public boolean updateData(int id, String newContent) { 
        String cacheKey = "data_" + id; 
        // 获取当前数据的版本号 
        DataRecord record = mockDatabaseGet(id); 
        if (record == null) { 
            return false; 
        } 
        int expectedVersion = record.getVersion();  
 
        try { 
            // 1. 第一次删除缓存 
            cache.remove(cacheKey);  
 
            // 2. 更新数据库（带版本校验） 
            boolean updateSuccess = mockDatabaseUpdate(id, newContent, expectedVersion); 
            if (!updateSuccess) { 
                return false; 
            } 
 
            // 3. 提交后安排延迟删除 
            scheduler.schedule(()  -> { 
                try { 
                    // 二次删除前的二次校验（可选） 
                    DataRecord current = mockDatabaseGet(id); 
                    if (current != null && current.getVersion()  > expectedVersion) { 
                        cache.remove(cacheKey);  // 只删除旧版本缓存 
                    } 
                } catch (Exception e) { 
                    // 处理异常，可添加重试逻辑 
                    e.printStackTrace();  
                } 
            }, 1, TimeUnit.SECONDS); // 延迟时间根据主从同步时间调整 
 
            return true; 
        } catch (Exception e) { 
            // 处理异常，可添加补偿逻辑 
            e.printStackTrace();  
            return false; 
        } 
    } 
 
    public static void main(String[] args) { 
        DataService service = new DataService(); 
        // 初始化数据 
        service.initData(1,  "Initial Content"); 
        // 获取数据 
        System.out.println("Initial  Data: " + service.getData(1));  
        // 更新数据 
        boolean result = service.updateData(1,  "Updated Content"); 
        System.out.println("Update  Result: " + result); 
        // 再次获取数据 
        System.out.println("Updated  Data: " + service.getData(1));  
    } 
}

要点

  1. mockDatabaseUpdate 方法中，当更新数据时，会先检查传入的期望版本号与数据库中记录的版本号是否一致。如果一致，会更新数据内容并将版本号加 1。
  2. getData 方法会先从缓存中查找数据，如果缓存中没有，则从数据库中获取数据，并将数据内容和版本号拼接后存入缓存。
  3. updateData 方法会先获取当前数据的版本号，然后执行延迟双删操作。在更新数据库时，会携带版本号进行校验，确保数据的一致性。

运行示例

在 main 方法中，我们演示了如何初始化数据、获取数据、更新数据和再次获取数据。运行程序后，你可以看到数据的初始状态、更新结果和更新后的数据。

通过这种方式，版本号和延迟双删机制可以协同工作，保证数据的一致性和缓存的正确性。

  • 延迟双删处理缓存层面的最终一致性
  • 第二次删除前的版本检查避免过度删除

典型时序：

  1. 请求A删除缓存
  2. 请求A更新数据库（版本2）
  3. 请求B读取缓存未命中，查询数据库（版本1）并填充缓存
  4. 延迟任务执行二次删除，发现数据库版本已更新，删除旧版本缓存
  5. 后续请求获取最新数据（版本2）并更新缓存

注意实际需要：

  • 替换mock数据库操作为真实DAO操作
  • 调整延迟时间（通常500ms-1s）
  • 添加缓存空值处理
  • 添加重试机制和监控

为什么要进行延迟双删

缓存和数据库数据的一致性一直是我们在后端开发中探讨的问题

先删除缓存再更新数据库情况

现在有两个线程

线程 1 是 写线程

线程 2 是 读线程

如果线程 1 是先删除缓存再更新数据库

在这个时间间隙 就是线程 1 写线程删除缓存和更行数据库的这个间隙

线程 2 读线程进来了

因为缓存已经被删除了 读线程尝试去数据库读取数据

脏数据就这样被写入了缓存

下次读的时候 因为缓存存在 所以一直读取的是旧数据

发生的几率比较大的原因往往是因为

更新数据库的数据是比较慢的

先更新数据库再删除缓存的情况

线程 1 是读线程 线程 1 首先去数据库读取到了旧数据

在写回缓存的这个间隙

线程 2 是写线程 更新了数据库为新数据

之后线程 1 才写入缓存

这样缓存里依旧是旧数据

但这种情况发生情况很小

应为缓存的写入很快

所以很难出现 读线程在写线程更改了数据库数据后再把数据写入缓存

而且另一种情况

线程 1 读线程 执行完毕后

线程 2 写线程 也最终会进行一次删除缓存的操作

思考

● 一种做法是在更新数据时也更新缓存，只是在更新缓存前先加一个分布式锁。因为这样在同一时间只允许一个线程更新缓存，就不会产生并发问题了。当然这么做对于写入的性能会有一些影响；

● 另一种做法同样也是在更新数据时更新缓存，只是给缓存加一个较短的过期时间。这样即使出现缓存不一致的情况，缓存的数据也会很快过期，对业务的影响也是可以接受。

延时双删实现

伪代码

# 延迟双删代码的实现

# 删除缓存
redis.delKey(X)

# 更新数据库
db.update(X)

# 睡眠
Thread.sleep(N)

# 再删除缓存
redis.delKey(X)

思考

在延迟双删策略中

我们需要在更新数据库之前

就先把缓存删掉

这样是为了防止在这个间隙有其他请求读取到了缓存

拿到的是失效的缓存数据

清除缓存后 在这个期间 其他请求是不会命中缓存的 会直接去数据库中读取最新数据

这样保证了数据的一致性和缓存的即时更新

在我看来延迟双删是在对比了先删除缓存再更新数据库还是先更新数据库的基础上 选择出了先更新数据库再删除缓存的基础上 的改进

更新数据库数据是一个很慢的过程

这样做可以高效的提高数据的一致性

再高并发读取的情况下 减轻数据库的读取压力 提高读取性能和响应速度

进一步优化

一、使用读写锁优化数据库并发控制

原理：通过区分读锁（共享锁）和写锁（排他锁），确保写操作期间独占资源，避免脏读和不可重复读问题。

示例场景：电商库存扣减

  1. 写锁应用：当用户下单扣减库存时，事务对库存记录加写锁（SELECT ... FOR UPDATE），阻止其他事务同时修改或读取未提交的库存数据。
  2. 读锁应用：商品详情页展示库存时，事务加读锁（SELECT ... LOCK IN SHARE MODE），允许其他读操作共享数据，但阻塞写操作。
  3. 效果：写锁独占期间，其他读请求需等待写锁释放，确保扣减操作的原子性，避免超卖。

二、高效缓存淘汰算法降低缓存失效影响

原理：通过动态调整缓存过期策略，减少因缓存集中失效导致的数据库瞬时压力。

示例场景：新闻热点数据缓存

  1. LRU算法优化：传统LRU可能误淘汰热点数据，可升级为 LRU-K（记录最近K次访问时间），优先保留高频访问数据。
  2. 时间窗口分散：为缓存键的过期时间添加随机值（如基础30分钟 + 随机0-10分钟），避免大量缓存同时失效引发雪崩。
  3. 主动更新机制：结合读写锁，在缓存失效前异步刷新数据（如后台线程检测过期前5分钟的热点Key，提前加载新数据）。

三、综合应用案例：社交平台评论系统

  1. 写锁控制评论发布
    • 用户发布评论时，事务对评论区数据加写锁，阻塞其他用户同时修改同一帖子，确保评论顺序和完整性。
    • 读锁允许其他用户持续加载已有评论，仅写操作短暂阻塞。
  1. LFU算法管理缓存
    • 使用 LFU（Least Frequently Used） 算法缓存热门帖子，自动淘汰低频访问的旧数据。
    • 结合 布隆过滤器 拦截无效查询（如已删除的帖子ID），减少缓存穿透。

四、注意事项

  1. 锁粒度选择：优先使用行级锁（如InnoDB的间隙锁）而非表锁，减少阻塞范围。
  2. 缓存一致性：采用 延迟双删策略（更新数据库后先删缓存，短暂延迟后再次删除），避免并发更新导致脏数据。
  3. 性能监控：通过工具（如Prometheus）监控锁等待时间和缓存命中率，动态调整锁策略和淘汰算法参数。

通过上述方法，可在高并发场景下平衡数据一致性与系统性能，减少因锁竞争或缓存失效导致的业务风险。

具体代码

我们现在要更新数据库

具体业务是插入数据

添加

 /**
     * 添加句子
     *
     * @param addSentenceDTO 注意提交是一个事务 如果失败则回滚 我们这边使用的是spring的事务框架
     */
    @Override
    @Transactional(rollbackFor = Exception.class, timeout = 10) // todo 如果插入标签过多 可能会导致事务回滚
    public void addSentenceWithTags(AddSentenceDTO addSentenceDTO) throws Exception {
        // 主记录插入
        AddSentenceReq addSentenceReq = addSentenceDTO.getAddSentenceReq();
        tSentencesMapper.addSentence(addSentenceReq);
        Long sentenceId = addSentenceReq.getSentenceId();
        // 关联标签插入
        List<AddTagsReq> tagsList = addSentenceDTO.getTagsList();
        AddSentenceTagReq addSentenceTagReq = new AddSentenceTagReq();
        addSentenceTagReq.setSentenceId(sentenceId);
        addSentenceTagReq.setTagsList(tagsList);
        int size = tagsList.size();
        if (size == 0) return;
        else {
            int i = tSentencesMapper.batchInsertTags(addSentenceTagReq); // 数据库插入标签并返回改变的标签数量
            if (i != size) {
                throw new Exception("传入了无效标签");
            }
        }
        // 此时已经更新了数据库 并且提交了事务(事务未回滚) 延迟双删 更新版本号
        TransactionSynchronizationManager.registerSynchronization(
                new TransactionSynchronization() {
                    @Override
                    public void afterCommit() {
                        DATA_VERSION.incrementAndGet(); // 版本号自增
                        String cacheKey = "balloonSentences:all" + DATA_VERSION;
                        delayDoubleDelete(cacheKey, 5, TimeUnit.SECONDS); // 执行延时双删
                        List<GetAllContentResp> dbData = tSentencesMapper.getAll(); // 更新elasticsearch
                        elasticsearchService.saveProduct(dbData);  // 写到elasticsearch里面去
                    }
                }
        );
    }

我们把代码逻辑进行了事务管理

当完成提交后

我们自增版本号

这边是使用的一个原子类

  // 原子类 版本号 这边表示的是当前数据版本的版本号
    private static final AtomicInteger DATA_VERSION = new AtomicInteger(1);

版本号机制重新构造缓存的 key

进行延迟双删

这边为什么又要有版本号机制又要进行双删

因为防止多个线程同时更新 所以要以最近的一次更新来刷新缓存

如果加锁的话 效率就会降低太多了

    /**
     * 更新缓存中全部句子的数据策略：延迟双删
     * 策略 先删除缓存 然后更新数据库 然后休眠 再删除缓存
     * 要求用分布式锁方式多线程进入操作数据库环境
     *
     * @param cacheKey
     * @param delay
     * @param unit
     */
    private void delayDoubleDelete(String cacheKey, int delay, TimeUnit unit) {
        RLock lock = redissonClient.getLock("lock:" + cacheKey);
        try {
            lock.lock();
            // 第一次删除（立即执行）
            redisService.deleteObject(cacheKey);
            // 延迟队列二次删除
            ScheduledExecutorService executor = Executors.newSingleThreadScheduledExecutor();
            executor.schedule(() -> {
                redisService.deleteObject(cacheKey);
                // 强制刷新缓存
                refreshCacheWithVersion(DATA_VERSION);
            }, delay, unit);
        } finally {
            lock.unlock();
        }
    }

之后再强制刷新缓存一遍

验证了我们刚才的想法

我们使用的要是最新的数据

缓存里面的也要是最新数据

/**
 * 强制刷新缓存
 *
 * @param currentVersion
 */
private void refreshCacheWithVersion(AtomicInteger currentVersion) {
    String cacheKey = "balloonSentences:all" + currentVersion;
    RLock lock = redissonClient.getLock("refresh:" + cacheKey);
    try {
        lock.lock();
        // 版本校验（防止旧版本覆盖）
        List<GetAllContentResp> newData = tSentencesMapper.getAll();
        // 删除缓存
        redisService.deleteObject(cacheKey);
        // 随机化TTL防雪崩 随机化过期时间
        redisService.setList(cacheKey, newData, RandomUtil.randomInt(30, 60), TimeUnit.MINUTES);
    } finally {
        lock.unlock();
    }
}

如何确定延时的时间

1.数据库性能

如果数据库更新快

可以选择较短的更新时间

2.缓存过期的时间

如果缓存过期的时间较长

可以选择缩短更新时间

以免过早的删除缓存导致数据不一致

思考

假设在延时双删策略中，第一次删除缓存后，会有一段时间的延时，然后再进行第二次删除缓存。如果此时缓存的过期时间设置得很短，比如只有几秒钟，那么在第二次删除缓存之前，缓存可能已经过期，而应用程序在读取缓存时会发现缓存已失效，从而不得不去数据库中查询最新数据。
为了避免这种情况，延时双删的延时时长应该要大于缓存的过期时间，确保在第二次删除缓存之前，缓存还是有效的，这样可以保证应用程序读取到的数据是一致的。
同时还需要考虑数据更新的频率和缓存的使用情况。如果数据更新较为频繁，那么延时双删的延时时长应该要适当缩短，以便及时更新缓存；如果缓存的使用率很低，可以适当延长延时时长，以减少对缓存服务的压力。


http://www.kler.cn/a/600891.html

相关文章：

  • 常用的几种思维方式
  • 2024年MathorCup数学建模C题物流网络分拣中心货量预测及人员排班解题全过程文档加程序
  • Android 12.0 WiFi连接默认设置静态IP地址功能实现
  • 【免费】2007-2019年各省地方财政非税收入数据
  • 【从零实现Json-Rpc框架】- 第三方库介绍 - Muduo篇
  • python每日十题(9)
  • 怎么用 DeepSeek 替你做外贸
  • C++设计模式-装饰模式：从基本介绍，内部原理、应用场景、使用方法，常见问题和解决方案进行深度解析
  • 【C语言】信号
  • Linux防火墙基础部分Firewalld防火墙(小白入门级别)
  • Linux 配置时间服务器
  • 2025 JMeter平替的五款工具
  • 高校校园交友微信小程序的设计与实现【lw+源码+部署+讲解】
  • 数据结构十五、排序
  • Unity 使用 Protobuf（Pb2）二进制数据全流程工具详解
  • Python（request库）
  • netty select/poll/epoll区别
  • CSS+JS 堆叠图片动态交互切换
  • 六级备考 词汇量积累（day11）
  • 最新DeepSeek-V3-0324：AI模型性能提升与新特性解析