MySQL与Redis数据同步实践与优化
一、数据不一致的典型场景
-
写入顺序不一致
当业务逻辑需要同时更新数据库和缓存时，若出现"先删缓存后更新DB"或"先更新DB后删缓存"操作失败，会导致缓存与数据库数据版本不一致。 -
并发读写冲突
高并发场景下可能出现：
- 线程A更新数据库
- 线程B读取旧缓存
- 线程A删除/更新缓存 此时缓存中残留旧数据
-
异步同步延迟
基于消息队列或binlog解析的异步同步方案，在网络波动或系统负载时可能出现同步延迟 -
缓存穿透/雪崩
恶意请求或突发流量导致：
- 缓存穿透：大量请求直接访问数据库
- 缓存雪崩：大量缓存同时过期
// 典型双写示例（问题代码）
public void updateProduct(Product product) {// 先更新数据库productDao.update(product); // 再更新缓存redisTemplate.opsForValue().set(product.getId(), product);
}
二、主流解决方案与Java实现
1. 延迟双删策略
public void updateProductWithDelay(Product product) {// 第一次删除缓存redisTemplate.delete(product.getId()); // 更新数据库productDao.update(product); // 延迟二次删除（使用异步线程）CompletableFuture.runAsync(() -> {try {Thread.sleep(1000); // 根据业务设置合理延迟redisTemplate.delete(product.getId());} catch (InterruptedException e) {Thread.currentThread().interrupt();}});
}
2. 基于Binlog的同步（Canal实现）
架构流程：
MySQL -> Canal Server -> Kafka -> 数据消费服务 -> Redis
// Canal客户端示例
@KafkaListener(topics = "canal_topic")
public void handleMessage(String message) {CanalMessage canalMsg = JSON.parseObject(message, CanalMessage.class);if ("UPDATE".equals(canalMsg.getType())) {canalMsg.getData().forEach(item -> {String key = "product:" + item.get("id");redisTemplate.opsForValue().set(key, item);});}
}
3. 分布式锁保障一致性
public Product getProduct(String id) {String cacheKey = "product:" + id;Product product = redisTemplate.opsForValue().get(cacheKey);if (product == null) {RLock lock = redissonClient.getLock("lock:" + cacheKey);try {lock.lock();// 双重检查锁product = redisTemplate.opsForValue().get(cacheKey);if (product == null) {product = productDao.findById(id);redisTemplate.opsForValue().set(cacheKey, product, 30, TimeUnit.MINUTES);}} finally {lock.unlock();}}return product;
}
三、优化实践方案
1. 异步批处理优化
// 使用Guava的批量收集器
@Bean
public BatchProcessor<DataChangeEvent> batchProcessor() {return BatchProcessor.create(events -> {List<RedisCommand> commands = events.stream().map(e -> new RedisCommand("SET", e.getKey(), e.getValue())).collect(Collectors.toList());redisTemplate.executePipelined(commands);},500, // 批量大小100, // 缓冲时间(ms)4 // 并发线程数);
}
2. 熔断降级策略
// 使用Resilience4j实现
CircuitBreaker circuitBreaker = CircuitBreaker.ofDefaults("redis");
RateLimiter rateLimiter = RateLimiter.of(100, Duration.ofSeconds(1));public Product getProductSafe(String id) {return Decorators.ofSupplier(() -> getProduct(id)).withCircuitBreaker(circuitBreaker).withRateLimiter(rateLimiter).withFallback(Exception.class, e -> productDao.findById(id)).get();
}
3. 数据版本控制
// 添加版本号字段
@Data
public class Product {private Long id;private String name;private Long version; // 数据版本
}// 更新时校验版本
public boolean updateWithVersion(Product product) {String luaScript = "if redis.call('get', KEYS[1]) == ARGV[1] then " +"redis.call('set', KEYS[1], ARGV[2]) return 1 else return 0 end";Long result = redisTemplate.execute(new DefaultRedisScript<>(luaScript, Long.class),Collections.singletonList("product:" + product.getId()),String.valueOf(product.getVersion() - 1),product.toString());return result == 1;
}
四、监控指标体系建设
-
关键监控指标：
- 同步延迟时间（Redis_Last_Update - DB_Update_Time）
- 缓存命中率（keyspace_hits / (keyspace_hits + keyspace_misses)）
- 同步失败率（failed_sync_count / total_sync_count）
-
日志追踪方案：
// 使用MDC实现请求链路追踪
public Product getProduct(String id) {MDC.put("traceId", UUID.randomUUID().toString());try {// 业务逻辑} finally {MDC.clear();}
}
五、总结与展望
通过组合使用延迟双删、binlog同步、分布式锁等方案，可构建不同一致性级别的同步系统。建议根据业务场景选择：
- 强一致性场景：分布式锁 + 同步双写
- 最终一致性场景：Canal + 消息队列
- 高性能场景：多级缓存 + 异步批处理