本文还是对《区块链：原理、设计与应用》的一个基础技术的总结和摘录。
散列的本质，是把任意内容，映射成固定长度的内容域里的某一个内容。
布隆过滤器的本质，是在常数时间内回答，“一个元素是否在一个集合内”的问题。
直观的方法及其缺陷
假设我们总可以把任意内容映射到某一个数组的 item 上，那么只要看看那个数组的 item 是否为空，就可以确认某一个内容是否存在。然而现实之中，一个数组总是会产生冲突，操作性能会因为局部冲突而产生退化。
多重散列的布隆过滤器
布隆过滤器的原理很简单，就是插入元素时，在一个容量为 m 的bit数组上， 用 k 要确认某个内容是否存在，可以使用布隆过滤器（Bloom Filter），它是一种高效的空间节省型数据结构，用于快速判断一个元素是否属于某个集合。以下是关键点总结：
布隆过滤器工作原理
- 插入元素：使用
k 个不同的哈希函数将元素映射到位数组（bit array）中的 k 个位置，并将这些位置标记为1。
- 查询元素：同样使用
k 个哈希函数查找对应的 k 个位置，如果所有位都是1，则认为该元素可能存在；如果至少有一个位是0，则该元素一定不存在。
特点
- 无假阴性：如果布隆过滤器说元素不存在，那它一定不在集合中。
- 可能有假阳性：如果布隆过滤器说元素存在，它可能实际上并不存在（误报）。
- 不支持删除（标准版本）：因为多个元素可能共享某些位，删除一个元素可能会误影响其他元素。但可以扩展为计数布隆过滤器（Counter Bloom Filter）支持删除。
优化参数
- 最佳哈希函数数量：$$k = (\ln 2) \cdot \frac{m}{n}$$，其中：
m 是位数组大小（bit数）n 是预期插入的元素数量
- 错误率越低，所需位数越多。
应用示例：找出多个大文件中的共同URL
- 当内存受限时（如4GB），布隆过滤器可以有效减少内存使用。
- 可先将一个文件的URL插入布隆过滤器，再逐个检查另一个文件的URL是否在其中。
- 若需处理多个文件，可使用多轮布隆过滤器或结合外部排序方法。
一个布隆过滤器相关的面试题
问题实例：给你A,B两个文件，各存放50亿条URL，每条URL占用64字节，内存限制是4G，让你找出A,B文件共同的URL。如果是三个乃至n个文件呢？
根据这个问题我们来计算下内存的占用，4G=2^32大概是40亿*8大概是340亿，n=50亿，如果按出错率0.01算需要的大概是650亿个bit。现在可用的是340亿，相差并不多，这样可能会使出错率上升些。另外如果这些urlip是一一对应的，就可以转换成ip，则大大简单了。
小结
布隆过滤器适用于需要快速判断元素是否存在、容忍一定误报率、但不能接受假阴性情况的场景，如缓存系统、数据库查询优化、网络应用等。 k 个 bit，而查找元素时，再用 k 个散列函数来寻找 k 个 bit，若这 k 个 bit 都被标记过了，则这个内容存在。
普通的布隆过滤器是不应该支持删除的，因为删除一个 bit 可能顺便删除掉其他内容的 bit。但如果把 bit 换成一个计数器，那么就可以考虑删除了。这也就会产生 counter bloom filter。
当hash函数个数k=(ln2)*(m/n)时错误率最小。
实际上，不管是散列算法，还是布隆过滤器，基本思想是一致的，都是基于内容的编址。Hash 函数存在冲突，布隆过滤器同样存在冲突。这就造成了两种方法都存在着假阳性误报问题（false positive），但绝不会存在假阴性漏报的问题（false negative）。布隆过滤器的误报率比单纯的散列算法低得多。
所以依据布隆过滤器设计方案的原则是：
- 把所有可能需要查存的数据一次性加载进过滤器里。
- 针对每个请求进行测试，如果测试不存在，则拒绝请求。
- 如果存在，需要走入实地查询流程。
- 合理选择参数（位数组大小、哈希函数个数、预期元素个数、误判率）。
- 最佳哈希函数数量：( k = \frac{m}{n} \times \ln(2) )
- 位数组大小：( m = -\frac{n \times \ln(p)}{(\ln(2))^2} )
- 误判率：( p \approx \left(1 - e^{-\frac{k \times n}{m}}\right)^k )
- 考虑误判率对业务的影响。对于一些对准确性要求不高的场景，可以接受较高的误判率以节省空间；而对于对准确性要求较高的场景，则需要降低误判率。
- 避免过度使用布隆过滤器。当数据量较小且内存足够时，使用HashSet可以避免误判问题。因此，在选择使用布隆过滤器时，需要权衡其优缺点。
- 考虑扩展性和维护性。在分布式系统中，布隆过滤器的状态可能需要在多个节点之间共享。此时，需要考虑如何高效地同步和更新布隆过滤器的状态。此外，布隆过滤器的维护也很重要，例如在数据更新时，需要及时更新布隆过滤器的状态，以保证其准确性。
实战
存穿透防护
场景描述
具体实现方案
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
| @Service
public class ProductCacheService {
private BloomFilter<String> productBloomFilter;
@Autowired
private RedisTemplate<String, Object> redisTemplate;
@PostConstruct
public void initBloomFilter() {
productBloomFilter = BloomFilter.create(
Funnels.stringFunnel(Charset.defaultCharset()),
1_000_000,
0.0001
);
List<String> existingProductIds = productRepository.getAllProductIds();
existingProductIds.forEach(productBloomFilter::put);
}
public Product getProduct(String productId) {
if (!productBloomFilter.mightContain(productId)) {
return null;
}
Product cached = (Product) redisTemplate.opsForValue()
.get("product:" + productId);
if (cached != null) {
return cached;
}
Product product = productRepository.findById(productId);
if (product != null) {
redisTemplate.opsForValue()
.set("product:" + productId, product, Duration.ofHours(1));
}
return product;
}
public void addProduct(Product product) {
productRepository.save(product);
productBloomFilter.put(product.getId());
}
}
|
性能提升
- ❌ 攻击前：恶意查询不存在的商品ID → 每次都查数据库
- ✅ 防护后：99.99%的恶意请求被布隆过滤器拦截
爬虫URL去重
场景描述
具体实现方案
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
| @Component
public class WebCrawler {
private BloomFilter<String> visitedUrls;
private Queue<String> crawlQueue = new LinkedBlockingQueue<>();
@PostConstruct
public void initCrawler() {
visitedUrls = BloomFilter.create(
Funnels.stringFunnel(Charset.defaultCharset()),
10_000_000,
0.001
);
}
public void addUrlToCrawl(String url) {
String normalizedUrl = normalizeUrl(url);
if (visitedUrls.mightContain(normalizedUrl)) {
log.debug("URL可能已访问，跳过: {}", normalizedUrl);
return;
}
crawlQueue.offer(normalizedUrl);
log.info("新URL加入队列: {}", normalizedUrl);
}
@Async
public void crawl() {
while (!crawlQueue.isEmpty()) {
String url = crawlQueue.poll();
try {
Document doc = Jsoup.connect(url).get();
visitedUrls.put(url);
Elements links = doc.select("a[href]");
for (Element link : links) {
String newUrl = link.absUrl("href");
addUrlToCrawl(newUrl);
}
processPageContent(doc);
} catch (Exception e) {
log.error("爬取失败: {}", url, e);
}
}
}
private String normalizeUrl(String url) {
return url.toLowerCase()
.replaceAll("#.*$", "")
.replaceAll("/$", "");
}
}
|
内存效率对比
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
| graph LR
subgraph "传统方法 HashSet"
A[1000万URL] --> B[每个URL平均100字节]
B --> C[需要约1GB内存]
end
subgraph "布隆过滤器"
D[1000万URL] --> E[误判率0.1%]
E --> F[仅需约14MB内存]
end
C --> G[内存使用减少98.6%]
F --> G
style G fill:#c8e6c9
|
分布式系统去重
场景描述
具体实现方案
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
| @Service
public class DistributedDeduplicationService {
@Autowired
private RedisTemplate<String, Object> redisTemplate;
private static final String BLOOM_FILTER_KEY = "dedup:bloom_filter";
private static final int EXPECTED_ELEMENTS = 10_000_000;
private static final double FALSE_POSITIVE_RATE = 0.001;
public boolean addAndCheck(String messageId) {
int[] hashes = getHashValues(messageId);
boolean isNew = false;
for (int hash : hashes) {
Boolean bitValue = redisTemplate.opsForValue()
.getBit(BLOOM_FILTER_KEY, hash);
if (!bitValue) {
isNew = true;
redisTemplate.opsForValue()
.setBit(BLOOM_FILTER_KEY, hash, true);
}
}
redisTemplate.expire(BLOOM_FILTER_KEY, Duration.ofHours(24));
return isNew;
}
private int[] getHashValues(String input) {
int hash1 = Hashing.murmur3_32().hashString(input, StandardCharsets.UTF_8).asInt();
int hash2 = Hashing.crc32().hashString(input, StandardCharsets.UTF_8).asInt();
int[] hashes = new int[3];
for (int i = 0; i < 3; i++) {
hashes[i] = Math.abs((hash1 + i * hash2) % getBitArraySize());
}
return hashes;
}
private int getBitArraySize() {
return (int) (-EXPECTED_ELEMENTS * Math.log(FALSE_POSITIVE_RATE) / (Math.log(2) * Math.log(2)));
}
}
@Component
public class MessageProcessor {
@Autowired
private DistributedDeduplicationService dedupService;
@EventListener
public void handleMessage(MessageEvent event) {
String messageId = event.getMessageId();
if (!dedupService.addAndCheck(messageId)) {
log.warn("重复消息，已丢弃: {}", messageId);
return;
}
processNewMessage(event);
}
}
|
数据库查询优化
场景描述
具体实现方案
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
| @Repository
public class PartitionedUserRepository {
private Map<String, BloomFilter<String>> partitionBloomFilters = new HashMap<>();
@PostConstruct
public void initPartitionFilters() {
List<String> partitions = Arrays.asList("user_2023", "user_2024", "user_2025");
for (String partition : partitions) {
BloomFilter<String> filter = BloomFilter.create(
Funnels.stringFunnel(Charset.defaultCharset()),
1_000_000,
0.01
);
List<String> userIds = getUserIdsFromPartition(partition);
userIds.forEach(filter::put);
partitionBloomFilters.put(partition, filter);
}
}
public User findUserById(String userId) {
List<String> candidatePartitions = new ArrayList<>();
for (Map.Entry<String, BloomFilter<String>> entry : partitionBloomFilters.entrySet()) {
if (entry.getValue().mightContain(userId)) {
candidatePartitions.add(entry.getKey());
}
}
for (String partition : candidatePartitions) {
User user = queryUserFromPartition(partition, userId);
if (user != null) {
return user;
}
}
return null;
}
@Transactional
public void saveUser(User user) {
String partition = determinePartition(user);
saveUserToPartition(partition, user);
partitionBloomFilters.get(partition).put(user.getId());
}
}
|
各场景性能对比
这些实现方案展示了布隆过滤器在不同场景下的强大威力：
- 极低的内存开销
- 超快的查询速度
- 优秀的扩展性
- 显著的性能提升
