深度剖析Redis双写一致性问题的解决方案

// 典型问题代码示例
public void updateProduct(Product product) {
    // 操作1：更新数据库
    db.update(product); 
    // 操作2：删除缓存
    redis.del(product.getId());
}

public void updateWithDelayDelete(Product product) {
    // 第一阶段删除
    redis.delete(product.getId()); 
    
    // 数据库更新
    db.update(product);
    
    // 异步延时删除
    scheduledExecutor.schedule(() -> {
        redis.delete(product.getId());
    }, 500, TimeUnit.MILLISECONDS);
}

 // 读操作：使用读锁保证一致性
    public Integer getProductStock(Long productId) {
        String cacheKey = "product:stock:" + productId;
        RReadWriteLock lock = redissonClient.getReadWriteLock("product_lock:" + productId);
        
        try {
            // 1. 获取读锁（共享锁）
            lock.readLock().lock();
            
            // 2. 先查缓存
            Integer stock = (Integer) redisTemplate.opsForValue().get(cacheKey);
            if (stock != null) {
                return stock;
            }
            
            // 3. 缓存未命中，查数据库
            try {
                stock = jdbcTemplate.queryForObject(
                    "SELECT stock FROM product WHERE id = ?", 
                    Integer.class, 
                    productId
                );
            } catch (EmptyResultDataAccessException e) {
                return 0; // 处理数据不存在的情况
            }
            
            // 4. 写入缓存（设置过期时间防雪崩）
            redisTemplate.opsForValue().set(cacheKey, stock, 30, TimeUnit.MINUTES);
            return stock;
            
        } finally {
            // 5. 释放读锁
            lock.readLock().unlock();
        }
    }

    // 写操作：使用写锁保证强一致性
    public void updateProductStock(Long productId, int newStock) {
        String cacheKey = "product:stock:" + productId;
        RReadWriteLock lock = redissonClient.getReadWriteLock("product_lock:" + productId);
        
        try {
            // 1. 获取写锁（排他锁）
            lock.writeLock().lock();
            
            // 2. 更新数据库
            jdbcTemplate.update(
                "UPDATE product SET stock = ? WHERE id = ?", 
                newStock, 
                productId
            );
            
            // 3. 删除缓存（直接删除，下次读时重建）
            redisTemplate.delete(cacheKey);
            
        } finally {
            // 4. 释放写锁
            lock.writeLock().unlock();
        }
    }

// RocketMQ生产者
public void sendCacheUpdateMessage(String key) {
    Message message = new Message("CACHE_TOPIC", key.getBytes());
    rocketMQTemplate.send(message);
}

// RocketMQ消费者
@RocketMQMessageListener(topic = "CACHE_TOPIC")
public void processMessage(String key) {
    redis.delete(key);
    // 可选：重新加载最新数据
    Product product = db.get(key);
    redis.set(key, product);
}

# Canal服务端配置示例
canal:
  instance:
    master:
      address: 127.0.0.1:3306
    dbUsername: canal
    dbPassword: canal
    filter: .*\\..*