Redis 7.x 缓存与锁实战：基于外卖项目应对10个高频面试题的解决方案

Redis 7.x 缓存与锁实战：外卖系统高频面试题深度解析

在当今高并发的互联网应用中，缓存系统已成为架构设计的核心组件。Redis作为最受欢迎的内存数据库之一，其高性能、丰富的数据结构和灵活的扩展能力，使其成为解决系统性能瓶颈的利器。特别是在外卖这类实时性要求极高的业务场景中，Redis的应用直接关系到用户体验和系统稳定性。

本文将从一个典型的外卖系统架构出发，深入剖析Redis在实际业务中的关键应用场景，特别是那些在技术面试中频繁出现的"硬核"问题。不同于简单的概念罗列，我们会通过真实案例和可落地的代码方案，帮助开发者构建完整的知识体系，从容应对面试挑战。

1. Redis缓存异常场景与防御策略

1.1 缓存穿透：当查询"永远不存在"的数据

想象一下这样的场景：用户不断查询一个根本不存在的商品ID，每次请求都直接穿透缓存打到数据库。这种恶意攻击或程序bug导致的异常流量，可能瞬间压垮你的数据库。

解决方案不止是简单的"缓存空值"那么简单。我们需要构建多层次的防御体系：

public Product getProductById(String id) { // 1. 布隆过滤器预检查 if (!bloomFilter.mightContain(id)) { return null; } // 2. 查询缓存 String cacheKey = "product:" + id; String productJson = redisTemplate.opsForValue().get(cacheKey); // 3. 处理缓存命中 if (productJson != null) { if (productJson.equals("NULL")) { // 特殊标记的空值 return null; } return JSON.parseObject(productJson, Product.class); } // 4. 数据库查询 Product product = productMapper.selectById(id); // 5. 回填缓存 if (product == null) { redisTemplate.opsForValue().set(cacheKey, "NULL", 5, TimeUnit.MINUTES); } else { redisTemplate.opsForValue().set(cacheKey, JSON.toJSONString(product), 30, TimeUnit.MINUTES); } return product; }

关键点对比：

1.2 缓存击穿：热点Key突然失效的连锁反应

大促期间，某个爆款商品的缓存突然过期，瞬间涌入的请求直接冲垮数据库——这就是典型的缓存击穿。我们采用多级缓存+互斥锁的方案来应对：

public Product getHotProduct(String id) { // 第一级：本地缓存 Product product = localCache.get(id); if (product != null) { return product; } // 第二级：Redis缓存 String redisKey = "hot_product:" + id; String lockKey = "lock:" + redisKey; try { // 尝试获取分布式锁 boolean locked = redisTemplate.opsForValue().setIfAbsent(lockKey, "1", 10, TimeUnit.SECONDS); if (!locked) { // 未获取到锁，短暂等待后重试 Thread.sleep(100); return getHotProduct(id); } // 再次检查缓存（双检锁） String productJson = redisTemplate.opsForValue().get(redisKey); if (productJson != null) { return JSON.parseObject(productJson, Product.class); } // 数据库查询 product = productMapper.selectById(id); if (product != null) { // 更新多级缓存 redisTemplate.opsForValue().set(redisKey, JSON.toJSONString(product), 5, TimeUnit.MINUTES); localCache.put(id, product); } return product; } catch (InterruptedException e) { Thread.currentThread().interrupt(); throw new RuntimeException("获取商品信息中断", e); } finally { // 释放锁 redisTemplate.delete(lockKey); } }

提示：本地缓存建议使用Caffeine或Guava Cache，它们提供了更精细的过期策略和内存管理能力。

1.3 缓存雪崩：当大量Key同时失效

预防雪崩需要从多个维度进行设计：

1. 过期时间随机化：基础过期时间+随机偏移量

   int baseExpire = 30 * 60; // 30分钟 int randomExpire = ThreadLocalRandom.current().nextInt(0, 10 * 60); // 0-10分钟随机 redisTemplate.opsForValue().set(key, value, baseExpire + randomExpire, TimeUnit.SECONDS);

2. 多级缓存架构：

   • 本地缓存 → Redis集群 → 数据库
   • 各级缓存设置不同的过期策略

3. 熔断降级机制：

   @CircuitBreaker(fallbackMethod = "getProductFallback") public Product getProductWithCircuitBreaker(String id) { // 正常业务逻辑 } public Product getProductFallback(String id) { // 返回兜底数据或默认值 return defaultProduct; }

2. 数据一致性：缓存与数据库的同步艺术

2.1 经典问题：先更新数据库还是先删缓存？

两种主流方案的对比分析：

方案一：Cache-Aside Pattern

1. 更新数据库
2. 删除缓存

方案二：Write-Through Pattern

1. 更新缓存
2. 更新数据库

我们通过下表对比两种方案的优劣：

在外卖系统中，推荐采用延迟双删策略来平衡性能与一致性：

@Transactional public void updateProduct(Product product) { // 第一次删除缓存 redisTemplate.delete("product:" + product.getId()); // 更新数据库 productMapper.updateById(product); // 提交事务后异步延迟删除 CompletableFuture.runAsync(() -> { try { Thread.sleep(500); // 延迟500ms redisTemplate.delete("product:" + product.getId()); } catch (InterruptedException e) { Thread.currentThread().interrupt(); } }); }

2.2 监听Binlog的终极一致性方案

对于核心业务数据，可以采用更可靠的数据库变更监听方案：

@Bean public MysqlBinaryLogClient mysqlBinaryLogClient() { MysqlBinaryLogClient client = new MysqlBinaryLogClient(...); client.registerEventListener(event -> { if (event instanceof WriteRowsEventData) { // 解析变更数据 // 更新Redis缓存 } }); return client; }

这种方案的优点是完全解耦业务代码与缓存更新逻辑，但实现复杂度较高，适合对一致性要求极高的场景。

3. 分布式锁的进阶实践

3.1 商品超卖问题的多维度解决方案

面对秒杀场景下的超卖问题，我们对比几种常见方案：

Redis分布式锁的完整实现：

public boolean tryLock(String lockKey, long expireTime, TimeUnit timeUnit) { String lockId = UUID.randomUUID().toString(); // 使用Lua脚本保证原子性 String script = "if redis.call('setnx', KEYS[1], ARGV[1]) == 1 then " + "return redis.call('pexpire', KEYS[1], ARGV[2]) " + "else return 0 end"; boolean acquired = redisTemplate.execute( new DefaultRedisScript<>(script, Long.class), Collections.singletonList(lockKey), lockId, String.valueOf(timeUnit.toMillis(expireTime)) ) == 1; if (acquired) { // 启动看门狗线程续期 scheduleLockRenewal(lockKey, lockId, expireTime); } return acquired; } private void scheduleLockRenewal(String lockKey, String lockId, long expireTime) { ScheduledExecutorService executor = Executors.newSingleThreadScheduledExecutor(); executor.scheduleAtFixedRate(() -> { String script = "if redis.call('get', KEYS[1]) == ARGV[1] then " + "return redis.call('pexpire', KEYS[1], ARGV[2]) " + "else return 0 end"; redisTemplate.execute( new DefaultRedisScript<>(script, Long.class), Collections.singletonList(lockKey), lockId, String.valueOf(TimeUnit.SECONDS.toMillis(expireTime)) ); }, expireTime / 3, expireTime / 3, TimeUnit.SECONDS); }

3.2 RedLock算法与集群环境下的锁安全

在Redis集群环境下，简单的单节点锁可能失效。RedLock算法提供了更可靠的分布式锁实现：

public boolean tryRedLock(String lockKey, long expireTime, TimeUnit timeUnit) { String lockId = UUID.randomUUID().toString(); long endTime = System.currentTimeMillis() + timeUnit.toMillis(expireTime); int successCount = 0; for (RedisNode node : redisClusterNodes) { if (System.currentTimeMillis() > endTime) { break; } Jedis jedis = new Jedis(node.getHost(), node.getPort()); try { String result = jedis.set(lockKey, lockId, "NX", "PX", timeUnit.toMillis(expireTime)); if ("OK".equals(result)) { successCount++; } } finally { jedis.close(); } } // 必须在大多数节点上获取成功 boolean acquired = successCount >= (redisClusterNodes.size() / 2 + 1); if (acquired) { // 记录获取锁的节点信息用于释放 acquiredNodes.addAll(currentNodes.subList(0, successCount)); } return acquired; }

注意：RedLock实现成本较高，在非极端场景下，使用主从架构的Redis加上合理的锁超时设置通常已经足够。

4. Redis集群实战技巧

4.1 数据分片与热点发现

外卖系统中的商品数据访问往往遵循二八定律，我们需要识别并特殊处理热点数据：

// 热点Key检测器 public class HotKeyDetector { private final ConcurrentHashMap<String, AtomicLong> counterMap = new ConcurrentHashMap<>(); public void increment(String key) { counterMap.computeIfAbsent(key, k -> new AtomicLong()).incrementAndGet(); } public List<String> getHotKeys(int threshold) { return counterMap.entrySet().stream() .filter(entry -> entry.getValue().get() > threshold) .map(Map.Entry::getKey) .collect(Collectors.toList()); } } // 在业务代码中使用 @GetMapping("/product/{id}") public Product getProduct(@PathVariable String id) { hotKeyDetector.increment(id); // 正常业务逻辑 }

对于检测到的热点Key，可以采用本地缓存+多副本的策略：

public Product getProductWithHotspot(String id) { // 检查是否为热点 if (hotKeyDetector.isHotKey(id)) { // 使用本地缓存 Product product = localCache.get(id); if (product != null) { return product; } // 使用带随机后缀的Redis Key分散请求 int replica = ThreadLocalRandom.current().nextInt(0, 3); String redisKey = "product:" + id + ":" + replica; product = getFromRedis(redisKey); if (product != null) { localCache.put(id, product); return product; } } // 正常处理流程 return getProduct(id); }

4.2 集群配置与性能调优

Redis集群的关键配置项：

# redis.conf 关键配置 # 集群模式 cluster-enabled yes cluster-node-timeout 15000 # 内存管理 maxmemory 16gb maxmemory-policy volatile-lru # 持久化 appendonly yes appendfsync everysec # 连接池 maxclients 10000 tcp-keepalive 300

性能优化检查清单：

1. 连接池配置：

   @Bean public LettuceConnectionFactory redisConnectionFactory() { LettuceClientConfiguration config = LettuceClientConfiguration.builder() .commandTimeout(Duration.ofSeconds(1)) .clientOptions(ClientOptions.builder() .autoReconnect(true) .pingBeforeActivateConnection(true) .build()) .clientResources(ClientResources.builder() .ioThreadPoolSize(4) .computationThreadPoolSize(4) .build()) .build(); RedisStandaloneConfiguration serverConfig = new RedisStandaloneConfiguration("redis-host", 6379); return new LettuceConnectionFactory(serverConfig, config); }

2. Pipeline批量操作：

   List<Object> results = redisTemplate.executePipelined((RedisCallback<Object>) connection -> { for (Product product : products) { String key = "product:" + product.getId(); connection.stringCommands().set(key.getBytes(), serialize(product)); connection.expire(key.getBytes(), 3600); } return null; });

3. Lua脚本优化：

   -- 库存扣减脚本 local key = KEYS[1] local quantity = tonumber(ARGV[1]) local stock = tonumber(redis.call('get', key)) if stock >= quantity then redis.call('decrby', key, quantity) return 1 else return 0 end

在实际项目中，我们发现合理使用Pipeline可以将批量操作的性能提升5-10倍，而精心设计的Lua脚本则能减少网络往返带来的延迟。