Redis

关注公众号 jb51net

关闭
首页 > 数据库 > Redis > Redis缓存更新

Redis中6种缓存更新策略详解

作者：风象南

Redis作为一款高性能的内存数据库,已经成为缓存层的首选解决方案,然而,使用缓存时最大的挑战在于保证缓存数据与底层数据源的一致性,本文将介绍Redis中6种缓存更新策略,需要的朋友可以参考下

引言

Redis作为一款高性能的内存数据库，已经成为缓存层的首选解决方案。然而，使用缓存时最大的挑战在于保证缓存数据与底层数据源的一致性。缓存更新策略直接影响系统的性能、可靠性和数据一致性，选择合适的策略至关重要。

本文将介绍Redis中6种缓存更新策略。

策略一：Cache-Aside（旁路缓存）策略

工作原理

Cache-Aside是最常用的缓存模式，由应用层负责缓存和数据库的交互逻辑：

代码示例

@Service
public class UserServiceCacheAside {
    
    @Autowired
    private RedisTemplate<String, User> redisTemplate;
    
    @Autowired
    private UserRepository userRepository;
    
    private static final String CACHE_KEY_PREFIX = "user:";
    private static final long CACHE_EXPIRATION = 30; // 缓存过期时间（分钟）
    
    public User getUserById(Long userId) {
        String cacheKey = CACHE_KEY_PREFIX + userId;
        
        // 1. 查询缓存
        User user = redisTemplate.opsForValue().get(cacheKey);
        
        // 2. 缓存命中，直接返回
        if (user != null) {
            return user;
        }
        
        // 3. 缓存未命中，查询数据库
        user = userRepository.findById(userId).orElse(null);
        
        // 4. 将数据库结果写入缓存（设置过期时间）
        if (user != null) {
            redisTemplate.opsForValue().set(cacheKey, user, CACHE_EXPIRATION, TimeUnit.MINUTES);
        }
        
        return user;
    }
    
    public void updateUser(User user) {
        // 1. 先更新数据库
        userRepository.save(user);
        
        // 2. 再删除缓存
        String cacheKey = CACHE_KEY_PREFIX + user.getId();
        redisTemplate.delete(cacheKey);
        
        // 或者选择更新缓存
        // redisTemplate.opsForValue().set(cacheKey, user, CACHE_EXPIRATION, TimeUnit.MINUTES);
    }
}

优缺点分析

优点

缺点

适用场景

策略二：Read-Through（读穿透）策略

工作原理

Read-Through策略将缓存作为主要数据源的代理，由缓存层负责数据加载：

代码示例

首先定义缓存加载器接口：

public interface CacheLoader<K, V> {
    V load(K key);
}

实现Read-Through缓存管理器：

@Component
public class ReadThroughCacheManager<K, V> {
    
    @Autowired
    private RedisTemplate<String, V> redisTemplate;
    
    private final ConcurrentHashMap<String, CacheLoader<K, V>> loaders = new ConcurrentHashMap<>();
    
    public void registerLoader(String cachePrefix, CacheLoader<K, V> loader) {
        loaders.put(cachePrefix, loader);
    }
    
    public V get(String cachePrefix, K key, long expiration, TimeUnit timeUnit) {
        String cacheKey = cachePrefix + key;
        
        // 1. 查询缓存
        V value = redisTemplate.opsForValue().get(cacheKey);
        
        // 2. 缓存命中，直接返回
        if (value != null) {
            return value;
        }
        
        // 3. 缓存未命中，通过加载器获取数据
        CacheLoader<K, V> loader = loaders.get(cachePrefix);
        if (loader == null) {
            throw new IllegalStateException("No cache loader registered for prefix: " + cachePrefix);
        }
        
        // 使用加载器从数据源加载数据
        value = loader.load(key);
        
        // 4. 将加载的数据存入缓存
        if (value != null) {
            redisTemplate.opsForValue().set(cacheKey, value, expiration, timeUnit);
        }
        
        return value;
    }
}

使用示例：

@Service
public class UserServiceReadThrough {
    
    private static final String CACHE_PREFIX = "user:";
    private static final long CACHE_EXPIRATION = 30;
    
    @Autowired
    private ReadThroughCacheManager<Long, User> cacheManager;
    
    @Autowired
    private UserRepository userRepository;
    
    @PostConstruct
    public void init() {
        // 注册用户数据加载器
        cacheManager.registerLoader(CACHE_PREFIX, this::loadUserFromDb);
    }
    
    private User loadUserFromDb(Long userId) {
        return userRepository.findById(userId).orElse(null);
    }
    
    public User getUserById(Long userId) {
        // 直接通过缓存管理器获取数据，缓存逻辑由管理器处理
        return cacheManager.get(CACHE_PREFIX, userId, CACHE_EXPIRATION, TimeUnit.MINUTES);
    }
}

优缺点分析

优点

缺点

适用场景

策略三：Write-Through（写穿透）策略

工作原理

Write-Through策略由缓存层同步更新底层数据源：

代码示例

首先定义写入接口：

public interface CacheWriter<K, V> {
    void write(K key, V value);
}

实现Write-Through缓存管理器：

@Component
public class WriteThroughCacheManager<K, V> {
    
    @Autowired
    private RedisTemplate<String, V> redisTemplate;
    
    private final ConcurrentHashMap<String, CacheWriter<K, V>> writers = new ConcurrentHashMap<>();
    
    public void registerWriter(String cachePrefix, CacheWriter<K, V> writer) {
        writers.put(cachePrefix, writer);
    }
    
    public void put(String cachePrefix, K key, V value, long expiration, TimeUnit timeUnit) {
        String cacheKey = cachePrefix + key;
        
        // 1. 获取对应的缓存写入器
        CacheWriter<K, V> writer = writers.get(cachePrefix);
        if (writer == null) {
            throw new IllegalStateException("No cache writer registered for prefix: " + cachePrefix);
        }
        
        // 2. 同步写入数据库
        writer.write(key, value);
        
        // 3. 更新缓存
        redisTemplate.opsForValue().set(cacheKey, value, expiration, timeUnit);
    }
}

使用示例：

@Service
public class UserServiceWriteThrough {
    
    private static final String CACHE_PREFIX = "user:";
    private static final long CACHE_EXPIRATION = 30;
    
    @Autowired
    private WriteThroughCacheManager<Long, User> cacheManager;
    
    @Autowired
    private UserRepository userRepository;
    
    @PostConstruct
    public void init() {
        // 注册用户数据写入器
        cacheManager.registerWriter(CACHE_PREFIX, this::saveUserToDb);
    }
    
    private void saveUserToDb(Long userId, User user) {
        userRepository.save(user);
    }
    
    public void updateUser(User user) {
        // 通过缓存管理器更新数据，会同步更新数据库和缓存
        cacheManager.put(CACHE_PREFIX, user.getId(), user, CACHE_EXPIRATION, TimeUnit.MINUTES);
    }
}

优缺点分析

优点

缺点

适用场景

策略四：Write-Behind（写回）策略

工作原理

Write-Behind策略将写操作异步化处理：

代码示例

实现异步写入队列和处理器：

@Component
public class WriteBehindCacheManager<K, V> {
    
    @Autowired
    private RedisTemplate<String, V> redisTemplate;
    
    private final BlockingQueue<CacheUpdate<K, V>> updateQueue = new LinkedBlockingQueue<>();
    private final ConcurrentHashMap<String, CacheWriter<K, V>> writers = new ConcurrentHashMap<>();
    
    public void registerWriter(String cachePrefix, CacheWriter<K, V> writer) {
        writers.put(cachePrefix, writer);
    }
    
    @PostConstruct
    public void init() {
        // 启动异步写入线程
        Thread writerThread = new Thread(this::processWriteBehindQueue);
        writerThread.setDaemon(true);
        writerThread.start();
    }
    
    public void put(String cachePrefix, K key, V value, long expiration, TimeUnit timeUnit) {
        String cacheKey = cachePrefix + key;
        
        // 1. 更新缓存
        redisTemplate.opsForValue().set(cacheKey, value, expiration, timeUnit);
        
        // 2. 将更新放入队列，等待异步写入数据库
        updateQueue.offer(new CacheUpdate<>(cachePrefix, key, value));
    }
    
    private void processWriteBehindQueue() {
        List<CacheUpdate<K, V>> batch = new ArrayList<>(100);
        
        while (true) {
            try {
                // 获取队列中的更新，最多等待100ms
                CacheUpdate<K, V> update = updateQueue.poll(100, TimeUnit.MILLISECONDS);
                
                if (update != null) {
                    batch.add(update);
                }
                
                // 继续收集队列中可用的更新，最多收集100个或等待200ms
                updateQueue.drainTo(batch, 100 - batch.size());
                
                if (!batch.isEmpty()) {
                    // 按缓存前缀分组批量处理
                    Map<String, List<CacheUpdate<K, V>>> groupedUpdates = batch.stream()
                            .collect(Collectors.groupingBy(CacheUpdate::getCachePrefix));
                    
                    for (Map.Entry<String, List<CacheUpdate<K, V>>> entry : groupedUpdates.entrySet()) {
                        String cachePrefix = entry.getKey();
                        List<CacheUpdate<K, V>> updates = entry.getValue();
                        
                        CacheWriter<K, V> writer = writers.get(cachePrefix);
                        if (writer != null) {
                            // 批量写入数据库
                            for (CacheUpdate<K, V> u : updates) {
                                try {
                                    writer.write(u.getKey(), u.getValue());
                                } catch (Exception e) {
                                    // 处理异常，可以重试或记录日志
                                    log.error("Failed to write-behind for key {}: {}", u.getKey(), e.getMessage());
                                }
                            }
                        }
                    }
                    
                    batch.clear();
                }
                
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                break;
            } catch (Exception e) {
                log.error("Error in write-behind process", e);
            }
        }
    }
    
    @Data
    @AllArgsConstructor
    private static class CacheUpdate<K, V> {
        private String cachePrefix;
        private K key;
        private V value;
    }
}

使用示例：

@Service
public class UserServiceWriteBehind {
    
    private static final String CACHE_PREFIX = "user:";
    private static final long CACHE_EXPIRATION = 30;
    
    @Autowired
    private WriteBehindCacheManager<Long, User> cacheManager;
    
    @Autowired
    private UserRepository userRepository;
    
    @PostConstruct
    public void init() {
        // 注册用户数据写入器
        cacheManager.registerWriter(CACHE_PREFIX, this::saveUserToDb);
    }
    
    private void saveUserToDb(Long userId, User user) {
        userRepository.save(user);
    }
    
    public void updateUser(User user) {
        // 更新仅写入缓存，异步写入数据库
        cacheManager.put(CACHE_PREFIX, user.getId(), user, CACHE_EXPIRATION, TimeUnit.MINUTES);
    }
}

优缺点分析

优点

缺点

适用场景

策略五：刷新过期（Refresh-Ahead）策略

工作原理

Refresh-Ahead策略预测性地在缓存过期前进行更新：

代码示例

@Component
public class RefreshAheadCacheManager<K, V> {
    
    @Autowired
    private RedisTemplate<String, Object> redisTemplate;
    
    @Autowired
    private ThreadPoolTaskExecutor refreshExecutor;
    
    private final ConcurrentHashMap<String, CacheLoader<K, V>> loaders = new ConcurrentHashMap<>();
    
    // 刷新阈值，当过期时间剩余不足阈值比例时触发刷新
    private final double refreshThreshold = 0.75; // 75%
    
    public void registerLoader(String cachePrefix, CacheLoader<K, V> loader) {
        loaders.put(cachePrefix, loader);
    }
    
    @SuppressWarnings("unchecked")
    public V get(String cachePrefix, K key, long expiration, TimeUnit timeUnit) {
        String cacheKey = cachePrefix + key;
        
        // 1. 获取缓存项和其TTL
        V value = (V) redisTemplate.opsForValue().get(cacheKey);
        Long ttl = redisTemplate.getExpire(cacheKey, TimeUnit.MILLISECONDS);
        
        if (value != null) {
            // 2. 如果缓存存在但接近过期，触发异步刷新
            if (ttl != null && ttl > 0) {
                long expirationMs = timeUnit.toMillis(expiration);
                if (ttl < expirationMs * (1 - refreshThreshold)) {
                    refreshAsync(cachePrefix, key, cacheKey, expiration, timeUnit);
                }
            }
            return value;
        }
        
        // 3. 缓存不存在，同步加载
        return loadAndCache(cachePrefix, key, cacheKey, expiration, timeUnit);
    }
    
    private void refreshAsync(String cachePrefix, K key, String cacheKey, long expiration, TimeUnit timeUnit) {
        refreshExecutor.execute(() -> {
            try {
                loadAndCache(cachePrefix, key, cacheKey, expiration, timeUnit);
            } catch (Exception e) {
                // 异步刷新失败，记录日志但不影响当前请求
                log.error("Failed to refresh cache for key {}: {}", cacheKey, e.getMessage());
            }
        });
    }
    
    private V loadAndCache(String cachePrefix, K key, String cacheKey, long expiration, TimeUnit timeUnit) {
        CacheLoader<K, V> loader = loaders.get(cachePrefix);
        if (loader == null) {
            throw new IllegalStateException("No cache loader registered for prefix: " + cachePrefix);
        }
        
        // 从数据源加载
        V value = loader.load(key);
        
        // 更新缓存
        if (value != null) {
            redisTemplate.opsForValue().set(cacheKey, value, expiration, timeUnit);
        }
        
        return value;
    }
}

使用示例：

@Service
public class ProductServiceRefreshAhead {
    
    private static final String CACHE_PREFIX = "product:";
    private static final long CACHE_EXPIRATION = 60; // 1小时
    
    @Autowired
    private RefreshAheadCacheManager<String, Product> cacheManager;
    
    @Autowired
    private ProductRepository productRepository;
    
    @PostConstruct
    public void init() {
        // 注册产品数据加载器
        cacheManager.registerLoader(CACHE_PREFIX, this::loadProductFromDb);
    }
    
    private Product loadProductFromDb(String productId) {
        return productRepository.findById(productId).orElse(null);
    }
    
    public Product getProduct(String productId) {
        return cacheManager.get(CACHE_PREFIX, productId, CACHE_EXPIRATION, TimeUnit.MINUTES);
    }
}

线程池配置

@Configuration
public class ThreadPoolConfig {
    
    @Bean
    public ThreadPoolTaskExecutor refreshExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(5);
        executor.setMaxPoolSize(20);
        executor.setQueueCapacity(100);
        executor.setThreadNamePrefix("cache-refresh-");
        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        executor.initialize();
        return executor;
    }
}

优缺点分析

优点

缺点

适用场景

策略六：最终一致性（Eventual Consistency）策略

工作原理

最终一致性策略基于分布式事件系统实现数据同步：

代码示例

首先定义数据变更事件：

@Data
@AllArgsConstructor
public class DataChangeEvent {
    private String entityType;
    private String entityId;
    private String operation; // CREATE, UPDATE, DELETE
    private String payload;   // JSON格式的实体数据
}

实现事件发布者：

@Component
public class DataChangePublisher {
    
    @Autowired
    private KafkaTemplate<String, DataChangeEvent> kafkaTemplate;
    
    private static final String TOPIC = "data-changes";
    
    public void publishChange(String entityType, String entityId, String operation, Object entity) {
        try {
            // 将实体序列化为JSON
            String payload = new ObjectMapper().writeValueAsString(entity);
            
            // 创建事件
            DataChangeEvent event = new DataChangeEvent(entityType, entityId, operation, payload);
            
            // 发布到Kafka
            kafkaTemplate.send(TOPIC, entityId, event);
        } catch (Exception e) {
            log.error("Failed to publish data change event", e);
            throw new RuntimeException("Failed to publish event", e);
        }
    }
}

实现事件消费者更新缓存：

@Component
@Slf4j
public class CacheUpdateConsumer {
    
    @Autowired
    private RedisTemplate<String, Object> redisTemplate;
    
    private static final long CACHE_EXPIRATION = 30;
    
    @KafkaListener(topics = "data-changes")
    public void handleDataChangeEvent(DataChangeEvent event) {
        try {
            String cacheKey = buildCacheKey(event.getEntityType(), event.getEntityId());
            
            switch (event.getOperation()) {
                case "CREATE":
                case "UPDATE":
                    // 解析JSON数据
                    Object entity = parseEntity(event.getPayload(), event.getEntityType());
                    // 更新缓存
                    redisTemplate.opsForValue().set(
                            cacheKey, entity, CACHE_EXPIRATION, TimeUnit.MINUTES);
                    log.info("Updated cache for {}: {}", cacheKey, event.getOperation());
                    break;
                    
                case "DELETE":
                    // 删除缓存
                    redisTemplate.delete(cacheKey);
                    log.info("Deleted cache for {}", cacheKey);
                    break;
                    
                default:
                    log.warn("Unknown operation: {}", event.getOperation());
            }
        } catch (Exception e) {
            log.error("Error handling data change event: {}", e.getMessage(), e);
            // 失败处理：可以将失败事件放入死信队列等
        }
    }
    
    private String buildCacheKey(String entityType, String entityId) {
        return entityType.toLowerCase() + ":" + entityId;
    }
    
    private Object parseEntity(String payload, String entityType) throws JsonProcessingException {
        // 根据实体类型选择反序列化目标类
        Class<?> targetClass = getClassForEntityType(entityType);
        return new ObjectMapper().readValue(payload, targetClass);
    }
    
    private Class<?> getClassForEntityType(String entityType) {
        switch (entityType) {
            case "User": return User.class;
            case "Product": return Product.class;
            // 其他实体类型
            default: throw new IllegalArgumentException("Unknown entity type: " + entityType);
        }
    }
}

使用示例：

@Service
@Transactional
public class UserServiceEventDriven {
    
    @Autowired
    private UserRepository userRepository;
    
    @Autowired
    private DataChangePublisher publisher;
    
    public User createUser(User user) {
        // 1. 保存用户到数据库
        User savedUser = userRepository.save(user);
        
        // 2. 发布创建事件
        publisher.publishChange("User", savedUser.getId().toString(), "CREATE", savedUser);
        
        return savedUser;
    }
    
    public User updateUser(User user) {
        // 1. 更新用户到数据库
        User updatedUser = userRepository.save(user);
        
        // 2. 发布更新事件
        publisher.publishChange("User", updatedUser.getId().toString(), "UPDATE", updatedUser);
        
        return updatedUser;
    }
    
    public void deleteUser(Long userId) {
        // 1. 从数据库删除用户
        userRepository.deleteById(userId);
        
        // 2. 发布删除事件
        publisher.publishChange("User", userId.toString(), "DELETE", null);
    }
}

优缺点分析

优点

缺点

适用场景

缓存更新策略选择指南

选择合适的缓存更新策略需要考虑以下因素：

1. 业务特性考量

业务特征推荐策略
读多写少Cache-Aside 或 Read-Through
写密集型Write-Behind
高一致性需求Write-Through
响应时间敏感Refresh-Ahead
分布式系统最终一致性

2. 资源限制考量

资源约束推荐策略
内存限制Cache-Aside（按需缓存）
数据库负载高Write-Behind（减轻写压力）
网络带宽受限Write-Behind 或 Refresh-Ahead

3. 开发复杂度考量

复杂度要求推荐策略
简单实现Cache-Aside
中等复杂度Read-Through 或 Write-Through
高复杂度但高性能Write-Behind 或 最终一致性

结论

缓存更新是Redis应用设计中的核心挑战，没有万能的策略适用于所有场景。根据业务需求、数据特性和系统资源，选择合适的缓存更新策略或组合多种策略才是最佳实践。

在实际应用中，可以根据不同数据的特性选择不同的缓存策略，甚至在同一个系统中组合多种策略，以达到性能和一致性的最佳平衡。

以上就是Redis中6种缓存更新策略详解的详细内容，更多关于Redis缓存更新的资料请关注脚本之家其它相关文章！

您可能感兴趣的文章:
阅读全文