今天咱们聊聊一个在高并发系统中非常关键的安全防护话题：如何防止恶意ID查询攻击。

恶意查询攻击的威胁

在我们的日常开发工作中，经常会遇到这样的攻击场景：

• 攻击者通过脚本不断查询不存在的用户ID，每次查询都直达数据库
• 恶意用户批量查询不存在的商品ID，消耗数据库资源
• 刷单机器人查询大量不存在的订单ID，试图探测系统漏洞
• 爬虫程序恶意查询系统中的各种资源ID

这类攻击虽然看似简单，但威力不容小觑。当攻击者使用大量不存在的ID进行查询时，由于缓存穿透，每次请求都会打到数据库，可能导致数据库压力过大甚至宕机。

布隆过滤器的解决方案

什么是布隆过滤器

布隆过滤器是一种概率型数据结构，它可以告诉你"某样东西一定不存在"或"可能存在"。它的工作原理类似于一个筛子，能够快速过滤掉明显不存在的数据请求。

为什么选择布隆过滤器

相比传统的缓存方案，布隆过滤器有以下优势：

• 空间效率高：比存储实际数据节省大量空间
• 查询速度快：O(k)时间复杂度，k为哈希函数个数
• 防穿透效果好：能有效过滤不存在的数据查询
• 误判率可控：通过调节参数控制误判率

核心实现方案

1. 布隆过滤器实现

@Component
public class BloomFilter<T> {
    
    private final BitSet bitSet;
    private final int bitSetSize;
    private final int expectedNumberOfItems;
    private final int numberOfHashFunctions;
    
    private final HashFunction[] hashFunctions;
    
    public BloomFilter(int expectedNumberOfItems, double falsePositiveProbability) {
        this.expectedNumberOfItems = expectedNumberOfItems;
        this.bitSetSize = (int) Math.ceil(expectedNumberOfItems * 
            Math.log(falsePositiveProbability) / Math.log(1.0 / (Math.pow(2.0, Math.log(2.0)))));
        
        this.numberOfHashFunctions = (int) Math.ceil(Math.log(2) * bitSetSize / expectedNumberOfItems);
        this.bitSet = new BitSet(bitSetSize);
        
        // 初始化哈希函数
        this.hashFunctions = new HashFunction[numberOfHashFunctions];
        for (int i = 0; i < numberOfHashFunctions; i++) {
            hashFunctions[i] = new SimpleHash(bitSetSize, i + 1);
        }
    }
    
    public void add(T element) {
        if (element != null) {
            for (HashFunction hashFunction : hashFunctions) {
                int position = hashFunction.hash(element.toString());
                bitSet.set(position);
            }
        }
    }
    
    public boolean contains(T element) {
        if (element == null) {
            return false;
        }
        
        for (HashFunction hashFunction : hashFunctions) {
            int position = hashFunction.hash(element.toString());
            if (!bitSet.get(position)) {
                return false; // 一定不存在
            }
        }
        
        return true; // 可能存在
    }
    
    // 简单哈希函数实现
    private static class SimpleHash {
        private int cap;
        private int seed;
        
        public SimpleHash(int cap, int seed) {
            this.cap = cap;
            this.seed = seed;
        }
        
        public int hash(String value) {
            int result = 0;
            int len = value.length();
            for (int i = 0; i < len; i++) {
                result = seed * result + value.charAt(i);
            }
            return (cap - 1) & result;
        }
    }
}

2. 本地缓存集成

@Service
public class SafeQueryService {
    
    @Autowired
    private BloomFilter<Long> userBloomFilter;
    
    @Autowired
    private RedisTemplate<String, Object> redisTemplate;
    
    @Autowired
    private UserRepository userRepository;
    
    public User getUserById(Long userId) {
        // 1. 首先使用布隆过滤器检查ID是否存在
        if (!userBloomFilter.contains(userId)) {
            // 布隆过滤器确定不存在，直接返回null
            return null;
        }
        
        // 2. 布隆过滤器认为可能存在，检查本地缓存
        String cacheKey = "user:" + userId;
        User cachedUser = (User) redisTemplate.opsForValue().get(cacheKey);
        if (cachedUser != null) {
            return cachedUser;
        }
        
        // 3. 缓存未命中，查询数据库
        User user = userRepository.findById(userId);
        if (user != null) {
            // 查询到数据，放入缓存
            redisTemplate.opsForValue().set(cacheKey, user, Duration.ofMinutes(30));
        } else {
            // 未查询到数据，设置空值缓存，防止缓存穿透
            redisTemplate.opsForValue().set(cacheKey, null, Duration.ofMinutes(5));
        }
        
        return user;
    }
}

3. 布隆过滤器预热

@Component
public class BloomFilterInitializer {
    
    @Autowired
    private BloomFilter<Long> userBloomFilter;
    
    @Autowired
    private UserRepository userRepository;
    
    @EventListener(ApplicationReadyEvent.class)
    public void initializeBloomFilter() {
        // 从数据库加载所有存在的ID到布隆过滤器
        List<Long> allUserIds = userRepository.findAllIds();
        
        for (Long userId : allUserIds) {
            userBloomFilter.add(userId);
        }
        
        System.out.println("布隆过滤器初始化完成，共加载 " + allUserIds.size() + " 个ID");
    }
}

高级优化策略

1. 多级布隆过滤器

@Service
public class MultiLevelBloomFilter {
    
    // 本地JVM级别的布隆过滤器（快速检查）
    private final BloomFilter<Long> localBloomFilter;
    
    // Redis中的布隆过滤器（集群共享）
    private final RedisBloomFilter redisBloomFilter;
    
    public MultiLevelBloomFilter() {
        this.localBloomFilter = new BloomFilter<>(100000, 0.01); // 本地过滤器
        this.redisBloomFilter = new RedisBloomFilter(); // Redis过滤器
    }
    
    public boolean mightExist(Long id) {
        // 先检查本地过滤器
        if (!localBloomFilter.contains(id)) {
            return false; // 本地过滤器确定不存在
        }
        
        // 本地过滤器认为可能存在，再检查Redis过滤器
        return redisBloomFilter.mightContain(id);
    }
    
    public void add(Long id) {
        localBloomFilter.add(id);
        redisBloomFilter.add(id);
    }
}

2. 动态更新机制

@Component
public class DynamicBloomFilterUpdater {
    
    @Autowired
    private BloomFilter<Long> bloomFilter;
    
    @EventListener
    public void handleUserCreated(UserCreatedEvent event) {
        // 用户创建事件，将新ID加入布隆过滤器
        bloomFilter.add(event.getUserId());
    }
    
    @EventListener
    public void handleUserDeleted(UserDeletedEvent event) {
        // 用户删除事件，重新初始化布隆过滤器（简单实现）
        // 更好的方案是使用支持删除的布隆过滤器变种，如Counting Bloom Filter
        reinitializeBloomFilter();
    }
    
    private void reinitializeBloomFilter() {
        // 重新从数据库加载所有ID
        // 这里可以考虑增量更新的方案
    }
}

3. 监控与告警

@Component
public class BloomFilterMonitor {
    
    private final MeterRegistry meterRegistry;
    
    // 布隆过滤器命中率监控
    private final Counter bloomFilterHitCounter;
    private final Counter bloomFilterMissCounter;
    
    public BloomFilterMonitor(MeterRegistry meterRegistry) {
        this.meterRegistry = meterRegistry;
        this.bloomFilterHitCounter = Counter.builder("bloom_filter.hit")
                .description("布隆过滤器命中次数")
                .register(meterRegistry);
        this.bloomFilterMissCounter = Counter.builder("bloom_filter.miss")
                .description("布隆过滤器未命中次数")
                .register(meterRegistry);
    }
    
    public boolean checkAndMonitor(Long id, BloomFilter<Long> filter) {
        boolean mightExist = filter.contains(id);
        
        if (mightExist) {
            bloomFilterHitCounter.increment();
        } else {
            bloomFilterMissCounter.increment();
        }
        
        return mightExist;
    }
}

Redis布隆过滤器集成

对于分布式环境，可以使用Redis的布隆过滤器模块：

@Component
public class RedisBloomFilter {
    
    @Autowired
    private RedisTemplate<String, Object> redisTemplate;
    
    private static final String BLOOM_FILTER_KEY = "bloom_filter:user_ids";
    
    public boolean mightContain(Long id) {
        // 使用Redis布隆过滤器命令
        try {
            return (Boolean) redisTemplate.execute(
                (RedisCallback<Boolean>) con -> 
                    con.executeCommand(
                        new CommandObject<>(RedisCommands.BF_EXISTS, 
                                          BLOOM_FILTER_KEY, String.valueOf(id))));
        } catch (Exception e) {
            // Redis布隆过滤器不可用时的降级处理
            return true; // 保守起见，认为可能存在
        }
    }
    
    public void add(Long id) {
        try {
            redisTemplate.execute(
                (RedisCallback<Void>) con ->
                    con.executeCommand(
                        new CommandObject<>(RedisCommands.BF_ADD, 
                                          BLOOM_FILTER_KEY, String.valueOf(id))));
        } catch (Exception e) {
            // 记录错误日志
        }
    }
}

最佳实践建议

1. 参数调优：根据实际数据量和误判率要求调整布隆过滤器参数
2. 缓存策略：结合布隆过滤器使用多级缓存策略
3. 监控告警：监控布隆过滤器的命中率和误判率
4. 定期重建：定期重建布隆过滤器以处理数据更新
5. 降级方案：设计Redis不可用时的降级策略

通过布隆过滤器与本地缓存的结合，我们可以有效防止恶意ID查询攻击，保护数据库免受不必要的压力。

以上就是本期分享的内容，希望对你有所帮助。更多技术干货，请关注服务端技术精选，我们下期再见！