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

Redis 与 MySQL 数据一致性保障方案

news 2025/6/8 5:31:17

在高并发场景下，Redis 作为缓存中间件与 MySQL 数据库配合使用时，数据一致性是一个关键挑战。本文将详细探讨如何保障 Redis 与 MySQL 的数据一致性，并结合 Java 代码实现具体方案。

数据不一致的原因分析

在分布式系统中，Redis 与 MySQL 的数据不一致主要由以下原因导致：

读写并发问题：多个线程同时进行读写操作时，可能导致数据在缓存和数据库中的状态不一致
更新策略不当：缓存更新策略选择不合理，如先删除缓存再更新数据库时可能出现并发问题
异常处理不足：更新过程中出现异常，导致缓存和数据库的更新操作未完成

缓存更新策略选择

常见的缓存更新策略有以下几种：

Cache-Aside Pattern（旁路缓存模式）
- 读操作：先读缓存，缓存不存在则读数据库并更新缓存
- 写操作：先更新数据库，再删除缓存
Read/Write Through Pattern（读写穿透模式）
- 应用程序只操作缓存，由缓存层负责数据库的读写
Write Behind Caching Pattern（写回模式）
- 写操作只更新缓存，由缓存层异步更新数据库

在实际应用中，Cache-Aside Pattern 是最常用的策略，下面将详细介绍其实现。

基于 Cache-Aside Pattern 的 Java 实现

以下是基于 Spring Boot 框架实现的 Cache-Aside Pattern 代码示例：

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 java.util.concurrent.TimeUnit;@Service
public class UserService {@Autowiredprivate UserRepository userRepository;@Autowiredprivate RedisTemplate<String, Object> redisTemplate;// 缓存键前缀private static final String CACHE_KEY_PREFIX = "user:";// 缓存过期时间（秒）private static final long CACHE_EXPIRE_TIME = 3600;/*** 查询用户（Cache-Aside Pattern读取实现）*/public User getUserById(Long userId) {// 1. 先从Redis中获取数据String cacheKey = CACHE_KEY_PREFIX + userId;User user = (User) redisTemplate.opsForValue().get(cacheKey);if (user != null) {return user;}// 2. Redis中不存在，从数据库中获取user = userRepository.findById(userId).orElse(null);if (user != null) {// 3. 将数据库结果写入RedisredisTemplate.opsForValue().set(cacheKey, user, CACHE_EXPIRE_TIME, TimeUnit.SECONDS);}return user;}/*** 更新用户（Cache-Aside Pattern写入实现）*/@Transactionalpublic User updateUser(User user) {// 1. 先更新数据库User updatedUser = userRepository.save(user);// 2. 删除缓存String cacheKey = CACHE_KEY_PREFIX + user.getId();redisTemplate.delete(cacheKey);return updatedUser;}/*** 删除用户（Cache-Aside Pattern删除实现）*/@Transactionalpublic void deleteUser(Long userId) {// 1. 先删除数据库记录userRepository.deleteById(userId);// 2. 删除缓存String cacheKey = CACHE_KEY_PREFIX + userId;redisTemplate.delete(cacheKey);}
}

解决并发问题的优化方案

上述实现中，在高并发场景下仍可能出现数据不一致问题，以下是几种优化方案：

延迟双删策略

@Transactional
public User updateUser(User user) {// 1. 先删除缓存String cacheKey = CACHE_KEY_PREFIX + user.getId();redisTemplate.delete(cacheKey);// 2. 更新数据库User updatedUser = userRepository.save(user);// 3. 延迟一段时间后再次删除缓存（异步执行）CompletableFuture.runAsync(() -> {try {// 等待一段时间，确保读请求全部完成Thread.sleep(100);redisTemplate.delete(cacheKey);} catch (InterruptedException e) {Thread.currentThread().interrupt();}});return updatedUser;
}

分布式锁机制

public User getUserById(Long userId) {String cacheKey = CACHE_KEY_PREFIX + userId;User user = (User) redisTemplate.opsForValue().get(cacheKey);if (user != null) {return user;}// 获取分布式锁RLock lock = redissonClient.getLock("user_cache_lock:" + userId);try {// 尝试获取锁，等待10秒，自动释放时间30秒boolean isLocked = lock.tryLock(10, 30, TimeUnit.SECONDS);if (isLocked) {// 再次检查缓存，避免重复查询数据库user = (User) redisTemplate.opsForValue().get(cacheKey);if (user != null) {return user;}// 查询数据库user = userRepository.findById(userId).orElse(null);if (user != null) {redisTemplate.opsForValue().set(cacheKey, user, CACHE_EXPIRE_TIME, TimeUnit.SECONDS);}}} catch (InterruptedException e) {Thread.currentThread().interrupt();} finally {// 释放锁if (lock.isHeldByCurrentThread()) {lock.unlock();}}return user;
}

最终一致性保障方案

对于一些对实时一致性要求不是特别高的场景，可以采用异步补偿机制保证最终一致性：

import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;@Service
public class CacheSyncService {@Autowiredprivate KafkaTemplate<String, Object> kafkaTemplate;// 发送缓存同步消息public void sendCacheSyncMessage(Long userId) {kafkaTemplate.send("cache_sync_topic", userId);}// 消费缓存同步消息@KafkaListener(topics = "cache_sync_topic")public void handleCacheSyncMessage(Long userId) {try {// 查询数据库最新数据User user = userRepository.findById(userId).orElse(null);// 更新缓存if (user != null) {String cacheKey = CACHE_KEY_PREFIX + userId;redisTemplate.opsForValue().set(cacheKey, user, CACHE_EXPIRE_TIME, TimeUnit.SECONDS);}} catch (Exception e) {// 记录异常日志，可添加重试机制log.error("处理缓存同步消息失败，userId: {}", userId, e);}}
}