Redis缓存穿透问题详解与解决方案

▶Redis核心
- ▶数据结构
- ▶内存管理
  - 内存分配策略
  - 淘汰算法
▶持久化
- ▶RDB机制
  - 快照生成原理
  - 文件格式解析
- ▶AOF机制
  - 命令追加策略
  - 重写过程分析
▶高可用
- ▶主从复制
  - SYNC原理
  - 增量复制
- ▶哨兵机制
  - 故障检测
  - 领导选举
▶高级特性
- ▶事务系统
  - ACID实现
  - WATCH原理
- ▶Lua脚本
  - 沙盒环境
  - 执行管道
▶实战问题
- ▶缓存问题
- ▶数据一致性
  - 读写一致性
  - 双写一致性
- ▶性能优化
  - 大key处理
  - 热点key优化

发布时间: 2025-04-20 09:09

↑

# Redis缓存穿透问题详解与解决方案

## 什么是缓存穿透

缓存穿透是指查询一个根本不存在的数据，由于缓存无法命中，该请求会穿透缓存直接查询数据库，如果有大量这样的请求，数据库就会承受巨大压力，甚至可能导致系统崩溃。

与缓存雪崩和缓存击穿不同，缓存穿透的特点是：

1. **查询的数据在数据库中不存在**
2. **每次请求都会直接访问数据库**
3. **无法通过自动填充缓存来解决**

## 缓存穿透的危害

缓存穿透可能带来以下严重后果：

1. **数据库负载过高**：大量无效查询绕过缓存直接访问数据库，增加数据库负载。

2. **系统响应变慢**：数据库负载增加导致查询响应时间延长，影响整体系统性能。

3. **资源浪费**：消耗系统资源处理无意义的查询请求。

4. **安全风险**：可能被恶意利用作为拒绝服务攻击的手段。

## 缓存穿透的常见原因

1. **业务误操作**：应用程序逻辑错误，频繁查询不存在的数据。

2. **恶意攻击**：攻击者故意构造不存在的数据ID进行查询，绕过缓存对数据库发起攻击。

3. **数据不一致**：缓存与数据库的数据不一致，导致缓存中不存在而数据库中存在的数据被频繁查询。

## 缓存穿透的解决方案

### 1. 布隆过滤器

布隆过滤器（Bloom Filter）是一种空间效率很高的概率型数据结构，用于判断一个元素是否在一个集合中。它可以用来快速判断一个key是否存在于数据库中，避免对数据库的无效查询。

```java
// Java示例 - 使用Guava实现布隆过滤器
import com.google.common.hash.BloomFilter;
import com.google.common.hash.Funnels;

public class UserService {
    
    // 创建布隆过滤器，预计存入100万个用户ID，误判率为0.01
    private BloomFilter<Long> bloomFilter = BloomFilter.create(
        Funnels.longFunnel(),
        1000000,
        0.01);
    
    @PostConstruct
    public void initBloomFilter() {
        // 系统启动时加载所有用户ID到布隆过滤器
        List<Long> allUserIds = userMapper.selectAllUserIds();
        for (Long userId : allUserIds) {
            bloomFilter.put(userId);
        }
    }
    
    public UserInfo getUserInfo(Long userId) {
        // 先判断用户ID是否可能存在
        if (!bloomFilter.mightContain(userId)) {
            // 布隆过滤器判断该ID不存在，直接返回空
            return null;
        }
        
        // 查询缓存
        String cacheKey = "user:" + userId;
        UserInfo userInfo = redisTemplate.opsForValue().get(cacheKey);
        if (userInfo != null) {
            return userInfo;
        }
        
        // 查询数据库
        userInfo = userMapper.selectById(userId);
        if (userInfo != null) {
            // 设置缓存
            redisTemplate.opsForValue().set(cacheKey, userInfo, 1, TimeUnit.HOURS);
        } else {
            // 数据库中也不存在，设置一个空值到缓存，避免下次继续查询数据库
            // 这里的过期时间应该设置得短一些
            redisTemplate.opsForValue().set(cacheKey, NULL_VALUE, 5, TimeUnit.MINUTES);
        }
        
        return userInfo;
    }
}
```

```go
// Go示例 - 使用bloom包实现布隆过滤器
import (
    "github.com/bits-and-blooms/bloom/v3"
)

var userBloomFilter *bloom.BloomFilter

func InitBloomFilter() {
    // 创建布隆过滤器，预计存入100万个用户ID，误判率为0.01
    userBloomFilter = bloom.NewWithEstimates(1000000, 0.01)
    
    // 加载所有用户ID
    userIds, err := db.QueryAllUserIds()
    if err != nil {
        log.Fatal("Failed to load user IDs:", err)
    }
    
    // 将所有用户ID添加到布隆过滤器
    for _, userId := range userIds {
        userIdStr := strconv.FormatInt(userId, 10)
        userBloomFilter.Add([]byte(userIdStr))
    }
}

func GetUserInfo(userId int64) (*UserInfo, error) {
    userIdStr := strconv.FormatInt(userId, 10)
    
    // 判断用户ID是否可能存在
    if !userBloomFilter.Test([]byte(userIdStr)) {
        // 布隆过滤器判断该ID不存在，直接返回空
        return nil, nil
    }
    
    // 查询缓存
    cacheKey := fmt.Sprintf("user:%d", userId)
    userInfoJson, err := redisClient.Get(ctx, cacheKey).Result()
    
    if err == nil {
        // 缓存命中
        if userInfoJson == "NULL" {
            // 空值标记，表示数据不存在
            return nil, nil
        }
        
        var userInfo UserInfo
        json.Unmarshal([]byte(userInfoJson), &userInfo)
        return &userInfo, nil
    }
    
    // 查询数据库
    userInfo, err := db.QueryUserInfo(userId)
    if err != nil {
        return nil, err
    }
    
    if userInfo != nil {
        // 设置缓存
        userInfoJson, _ := json.Marshal(userInfo)
        redisClient.Set(ctx, cacheKey, userInfoJson, time.Hour)
        return userInfo, nil
    } else {
        // 数据库中也不存在，设置一个空值到缓存
        redisClient.Set(ctx, cacheKey, "NULL", 5*time.Minute)
        return nil, nil
    }
}
```

**布隆过滤器的优缺点**：

- **优点**：空间效率高，查询速度快。
- **缺点**：有一定的误判率；不支持删除元素；需要提前加载数据。

### 2. 缓存空值

对于不存在的数据，在缓存中设置一个特殊的空值，避免每次查询都访问数据库：

```java
// Java示例
public UserInfo getUserInfo(Long userId) {
    // 查询缓存
    String cacheKey = "user:" + userId;
    String cacheValue = redisTemplate.opsForValue().get(cacheKey);
    
    // 判断是否为空值标记
    if ("NULL".equals(cacheValue)) {
        return null; // 数据不存在，直接返回
    }
    
    UserInfo userInfo = null;
    if (cacheValue != null) {
        userInfo = JSON.parseObject(cacheValue, UserInfo.class);
        return userInfo;
    }
    
    // 缓存未命中，查询数据库
    userInfo = userMapper.selectById(userId);
    
    if (userInfo != null) {
        // 数据存在，设置到缓存
        redisTemplate.opsForValue().set(cacheKey, JSON.toJSONString(userInfo), 1, TimeUnit.HOURS);
    } else {
        // 数据不存在，设置空值标记，过期时间较短
        redisTemplate.opsForValue().set(cacheKey, "NULL", 5, TimeUnit.MINUTES);
    }
    
    return userInfo;
}
```

**缓存空值的优缺点**：

- **优点**：实现简单，无需额外组件。
- **缺点**：可能会占用大量缓存空间；短期内仍可能有穿透问题。

### 3. 请求合并

对于高并发系统，可以合并对同一个key的多个请求，只让一个请求去查询数据库，其他请求等待结果：

```java
// Java示例 - 使用Guava Cache实现请求合并
import com.google.common.cache.Cache;
import com.google.common.cache.CacheBuilder;

public class UserService {
    
    // 用于存储正在进行中的请求
    private Cache<String, Future<UserInfo>> requestCache = CacheBuilder.newBuilder()
        .expireAfterWrite(3, TimeUnit.SECONDS)
        .build();
    
    public UserInfo getUserInfo(Long userId) throws Exception {
        String cacheKey = "user:" + userId;
        
        // 尝试从Redis缓存获取
        UserInfo userInfo = redisTemplate.opsForValue().get(cacheKey);
        if (userInfo != null || "NULL".equals(userInfo)) {
            return "NULL".equals(userInfo) ? null : userInfo;
        }
        
        // 合并请求处理
        Future<UserInfo> future = requestCache.get(cacheKey, () -> {
            // 创建异步任务查询数据库
            return executorService.submit(() -> {
                // 再次检查缓存，双重检查锁定
                UserInfo result = redisTemplate.opsForValue().get(cacheKey);
                if (result != null || "NULL".equals(result)) {
                    return "NULL".equals(result) ? null : result;
                }
                
                // 查询数据库
                result = userMapper.selectById(userId);
                
                // 设置缓存
                if (result != null) {
                    redisTemplate.opsForValue().set(cacheKey, result, 1, TimeUnit.HOURS);
                } else {
                    redisTemplate.opsForValue().set(cacheKey, "NULL", 5, TimeUnit.MINUTES);
                }
                
                return result;
            });
        });
        
        // 等待结果返回
        return future.get(200, TimeUnit.MILLISECONDS); // 设置超时时间
    }
}
```

### 4. 接口层限流

对于可能被恶意攻击的接口，实施限流措施，控制单个用户或IP的请求频率：

```java
// Java示例 - 使用Guava RateLimiter实现限流
import com.google.common.util.concurrent.RateLimiter;

@RestController
public class UserController {
    
    // 创建限流器，每秒允许10个请求
    private RateLimiter rateLimiter = RateLimiter.create(10.0);
    
    @GetMapping("/user/{userId}")
    public ResponseEntity<UserInfo> getUserInfo(@PathVariable Long userId) {
        // 尝试获取令牌，如果获取不到则拒绝请求
        if (!rateLimiter.tryAcquire()) {
            return ResponseEntity.status(HttpStatus.TOO_MANY_REQUESTS).build();
        }
        
        // 正常处理请求
        UserInfo userInfo = userService.getUserInfo(userId);
        if (userInfo == null) {
            return ResponseEntity.notFound().build();
        }
        return ResponseEntity.ok(userInfo);
    }
}
```

### 5. 数据一致性校验

定期检查缓存和数据库的一致性，修复不一致的数据：

```java
// Java示例 - 定时任务检查数据一致性
@Component
@EnableScheduling
public class DataConsistencyChecker {
    
    @Autowired
    private RedisTemplate<String, Object> redisTemplate;
    
    @Autowired
    private UserMapper userMapper;
    
    @Scheduled(cron = "0 0 2 * * ?") // 每天凌晨2点执行
    public void checkDataConsistency() {
        // 获取所有用户ID
        List<Long> allUserIds = userMapper.selectAllUserIds();
        
        // 更新布隆过滤器
        BloomFilter<Long> newBloomFilter = BloomFilter.create(
            Funnels.longFunnel(),
            allUserIds.size() * 2, // 预留空间
            0.01);
        
        for (Long userId : allUserIds) {
            newBloomFilter.put(userId);
            
            // 检查缓存中的数据是否与数据库一致
            String cacheKey = "user:" + userId;
            UserInfo cachedUser = redisTemplate.opsForValue().get(cacheKey);
            UserInfo dbUser = userMapper.selectById(userId);
            
            if (cachedUser == null && dbUser != null) {
                // 缓存缺失，更新缓存
                redisTemplate.opsForValue().set(cacheKey, dbUser, 1, TimeUnit.HOURS);
            } else if ("NULL".equals(cachedUser) && dbUser != null) {
                // 缓存标记为空但数据库存在，修复缓存
                redisTemplate.opsForValue().set(cacheKey, dbUser, 1, TimeUnit.HOURS);
            } else if (cachedUser != null && dbUser == null) {
                // 缓存存在但数据库不存在，删除缓存或设置为空
                redisTemplate.opsForValue().set(cacheKey, "NULL", 5, TimeUnit.MINUTES);
            }
        }
        
        // 替换旧的布隆过滤器
        this.bloomFilter = newBloomFilter;
    }
}
```

## 实践建议

1. **组合使用多种方案**：在实际应用中，通常需要组合使用多种方案来全面防御缓存穿透问题。例如，同时使用布隆过滤器和缓存空值。

2. **监控异常请求**：建立有效的监控系统，及时发现异常的访问模式和请求量激增的情况。

3. **定期更新防御机制**：定期更新布隆过滤器，清理过期的空值缓存，保持防御机制的有效性。

4. **业务层面优化**：在业务设计层面避免使用不存在的ID进行查询，例如使用白名单机制控制可查询的ID范围。

5. **安全防护**：对外部接口实施更严格的安全措施，如参数校验、身份认证等，防止恶意攻击。

## 总结

缓存穿透是一个常见但危害严重的缓存问题，它可能导致数据库负载过高甚至系统崩溃。通过布隆过滤器、缓存空值、请求合并、接口限流和数据一致性校验等多种技术手段，可以有效防止和缓解缓存穿透问题。

在实际应用中，应根据系统特点和业务需求，选择合适的防御策略，并通过持续监控和优化，确保系统的稳定性和性能。

元素码农