一、数据不一致的典型场景
写入顺序不一致
当业务逻辑需要同时更新数据库和缓存时，若出现"先删缓存后更新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 + 消息队列
- 高性能场景：多级缓存 + 异步批处理