倒排索引与加密倒排索引
文档索引构建与加密存储，说白了，就是回答两个问题。
第一个：你怎么在海量文档里一秒找到关键词？
第二个：你既想享受搜索引擎的便利，又不想泄露文档内容，怎么办？
我在2018年第一次做一个企业内部文档搜索系统时，就遇到了这两个问题。当时需求很明确：几百个同事的合同、方案、报价单要能搜，但HR的法务文件必须加密存储。嗯，今天这篇文章，我就带你把这个坑踩平。
4.1 倒排索引：搜索引擎的「内功心法」
你想想看，假如你有一万个文件想找某句话。最笨的方法是什么？
挨个打开，逐字比对。O(n)复杂度，一万次扫描。这方法在20个文件时还行，一万个你就疯了。
倒排索引的思路截然不同。它反过来看问题：
不是「文档里有什么词」，而是「这个词出现在哪些文档」。
核心定义：倒排索引（Inverted Index）是一个从单词（Term）到文档ID列表（Posting List）的映射结构。每个单词会附带一个位置列表（Position List），记录它在文档中的出现位置。
举个小例子，三个文档：
- Doc1: "I love cats"
- Doc2: "I love dogs"
- Doc3: "Dogs love cats"
构建出来的倒排索引结构大概长这样：
// 简化版倒排索引
{
"i": { "doc_freq": 2, "postings": [{ "doc_id": 1, "positions": [0] },
{ "doc_id": 2, "positions": [0] }] },
"love": { "doc_freq": 3, "postings": [{ "doc_id": 1, "positions": [1] },
{ "doc_id": 2, "positions": [1] },
{ "doc_id": 3, "positions": [1] }] },
"cats": { "doc_freq": 2, "postings": [{ "doc_id": 1, "positions": [2] },
{ "doc_id": 3, "positions": [2] }] },
"dogs": { "doc_freq": 2, "postings": [{ "doc_id": 2, "positions": [2] },
{ "doc_id": 3, "positions": [0] }] }
}| 组件 | 说明 | 我的建议 |
|---|---|---|
| 词典 | 所有不重复的单词集合，通常用哈希表或B树存储 | 用哈希表，查询O(1) |
| 文档频率 | 包含该单词的文档总数 | 用于计算TF-IDF权重 |
| 倒排表 | 每个文档ID及其出现位置的链表 | 压缩存储，节省空间 |
| 元数据 | 文档长度、权重、更新时间等 | 用于查询排序 |
小技巧：实际项目中，倒排表不会用一个简单的数组。我会用 变长编码（Varint）+ 差值编码（Delta Encoding） 来压缩存储。比如文档ID序列[1, 3, 5, 10]会变成[1, 2, 2, 5]，再配合Varint编码，存储成本下降40%以上。
4.2 构建流程：从原始文档到可搜索索引
我习惯把一个文档处理成可搜索索引，分成四个步骤。每步都是坑，我一个个给你说。
- 文本清洗：去掉HTML标签、特殊字符、标点符号
- 分词（Tokenization）：把句子切成单词——英文空格切，中文需要分词器，比如jieba或HanLP
- 归一化（Normalization）：转小写、去掉词尾复数、提取词干（Stemming）
- 去停用词（Stop Words Removal）：去掉"The"、"is"、"at"等高频无意义词
来个实战代码。我用Python手写一个简易版：
import re
from collections import defaultdict
class SimpleInvertedIndex:
def __init__(self):
self.index = defaultdict(list)
self.stop_words = {'the', 'is', 'at', 'which', 'on', 'a', 'an', 'and', 'or'}
def tokenize(self, text):
# 1. 转小写
text = text.lower()
# 2. 去标点
text = re.sub(r'[^\w\s]', '', text)
# 3. 切词
tokens = text.split()
# 4. 去停用词
return [t for t in tokens if t not in self.stop_words]
def add_document(self, doc_id, text):
tokens = self.tokenize(text)
for pos, in enumerate(tokens):
self.index[].append((doc_id, pos))
def search(self, query):
tokens = self.tokenize(query)
# 返回所有包含所有查询词的文档ID交集
if not tokens:
return []
result_set = set(doc_id for doc_id, _ in self.index[tokens[0]])
for in tokens[1:]:
doc_ids = set(doc_id for doc_id, _ in self.index.get(, []))
result_set &= doc_ids
return list(result_set)
# 使用示例
idx = SimpleInvertedIndex()
idx.add_document(1, "The cat sat on the mat")
idx.add_document(2, "Dogs play in the park")
print(idx.search("cat mat")) # 输出: [1]
print(idx.search("park")) # 输出: [2]我曾经踩过的坑：千万别忘了做分词后的排序。如果倒排表里的文档ID不是有序的，做交集查询时复杂度会是O(n²)。我当年第一次上线时忘记排了，结果三个查询就把CPU打到100%。
4.3 倒排索引的存储结构
索引建完了，存在哪？常见的四种方案：
| 方案 | 适用场景 | 优点 | 缺点 |
|---|---|---|---|
| 内存哈希表 | 小型系统、原型开发 | 极快查询、实现简单 | 数据不可靠、内存上限 |
| B+树（磁盘） | 传统数据库、Lucene | 持久化、支持范围查询 | 写放大 |
| LSM-Tree | Elasticsearch、RocksDB | 写吞吐极高 | 读放大、合并开销 |
| 压缩位图（Roaring Bitmaps） | 倒排表交集/并集运算 | 极致压缩、集合运算快 | 动态更新复杂 |
我个人的经验是：如果你是在做密文检索接口，最常用的是内存哈希表+序列化到磁盘的组合。既要快，又要有持久化。毕竟加密后的索引内容，你不能随便丢。
4.4 加密倒排索引
好了，到这里一切都很正常。但问题来了——如果这些文档是机密文件呢？里面的关键词比如「核潜艇」、「并购A公司」不能明文存？
你想想看，你不能把索引直接扔到数据库里，因为DB管理员会看到。你也不能把索引加密成密文再存，因为那就没法搜了。
秘密就在加密方式上。
4.5 两种主流的加密倒排索引方案
- 方案一：基于确定型加密
- 用AES-SIV或PRP加密每个关键词。
- 相同关键词始终加密为相同密文。
- 优点：查询时完全兼容原倒排索引结构，搜得快。
- 缺点：频率泄露——攻击者可以通过密文频率反推明文。
- 方案二：基于布隆过滤器+双哈希
- 为每个文档构建一个加密布隆过滤器。
- 查询时把关键词哈希后，用加密令牌过滤。
- 优点：抗频率攻击，一次查询返回伪造结果干扰。
- 缺点：有假阳性概率，索引规模较小时有误报。
我的建议：如果你对安全等级要求高——比如医疗档案、法律文书，用方案二。如果你只想要「快速可用」，比如内部知识库，用方案一足够了。
4.6 实战：加密倒排索引的构建核心代码
我们来写一个基于确定型加密的简单实现。我直接贴出加密索引构建和查询的核心逻辑：
import os
from cryptography.fernet import Fernet
from collections import defaultdict
class EncryptedIndex:
def __init__(self, key=None):
# 如果没有传入密钥，自动生成一个
self.key = key or Fernet.generate_key()
self.cipher = Fernet(self.key)
self.encrypted_index = defaultdict(list) # 加密词 -> 加密的(文档ID, 位置)
def _encrypt_term(self, term):
"""将明文字段加密为确定性的密文"""
return self.cipher.encrypt(term.encode())
def _encrypt_posting(self, doc_id, pos):
"""将文档ID和位置加密后拼装"""
raw = f"{doc_id}:{pos}".encode()
return self.cipher.encrypt(raw)
def add_document(self, doc_id, tokens):
"""构建加密倒排索引"""
for pos, in enumerate(tokens):
enc_term = self._encrypt_term()
enc_posting = self._encrypt_posting(doc_id, pos)
self.encrypted_index[enc_term].append(enc_posting)
def search(self, query_tokens):
"""查询——加密关键词，然后取交集"""
enc_tokens = [self._encrypt_term(t) for t in query_tokens]
if not enc_tokens:
return []
# 假设第一个词的结果集为基础
result_set = set(self.encrypted_index.get(enc_tokens[0], []))
for enc_ in enc_tokens[1:]:
current_set = set(self.encrypted_index.get(enc_, []))
result_set &= current_set
# 解密结果返回
decrypted = []
for enc_doc_pos in result_set:
plain = self.cipher.decrypt(enc_doc_pos).decode()
doc_id, pos = plain.split(":")
decrypted.append((int(doc_id), int(pos)))
return decrypted
# 模拟使用
key = Fernet.generate_key() # 实际应该安全存储
eidx = EncryptedIndex(key)
eidx.add_document(1, ["secure", "search", "project"])
eidx.add_document(2, ["search", "encryption"])
results = eidx.search(["search"])
print(f"Found {len(results)} results: {results}")注意：确定型加密的关键词密文是固定的。如果你的攻击者能看到两次时间点的索引快照，他能通过频率变化猜到关键词。我在一个日志管理项目里踩过这个坑，后来换成双布隆过滤器方案才解决。
4.7 核心流程图
下面这张图，展示从原始文档到加密倒排索引，再到客户端的完整构建流程。你自己仔细看，每条线都是有意义的。
4.8 避坑指南
做加密倒排索引工程落地，有三个地方最容易翻车：
- 索引同步时机：加密索引要定期刷新吗？我在项目中遇到过，用增量+全量双策略。每10分钟增量更新，每天凌晨全量重建一次，避免索引膨胀后查询效率下降。
- 密钥轮转：轮换密钥时，旧密文和新密文混存。解决方案是加一个索引版本号，查询时先判断版本，再用对应密钥解密。
- 假阳性问题：布隆过滤器在密文检索场景下最烦人。设置合理容量：插入数=文档数×平均词数×1.5，误判率控制在0.1%以内才可用。
我曾经翻车翻得很惨的一次：在生产环境中，我把所有加密倒排索引放在一个哈希表里。结果词多了，哈希冲突严重，查询全变成了扫描。后来改用分层滚动哈希表——按关键词首字母分桶，每个桶独立哈希，冲突率降到万分之一以下。
4.9 小结
这一章的核心就三点，你记住就好：
- 倒排索引是搜索引擎的根基，从词映射到文档ID。
- 构建时要做好分词、归一化、排序、倒排表压缩。
- 加密倒排索引把传统索引的每一个单元做确定性加密或布隆过滤器加密，从而在保证基本检索功能的同时，防止泄露文档内容和频率。