集合对比优化：大规模集合列表匹配性能提升方案

核心问题分析 #

原实现的时间复杂度为 O(MNK)（M是A的长度，N是B的长度，K为集合平均元素数），当M、N达到50万时，这种嵌套遍历的逻辑会产生天文级别的计算量，完全无法高效运行。必须从数据结构层面重构匹配逻辑，才能实现数量级的性能提升。

最优优化思路：构建元素到B集合的反向索引 #

通过预构建元素→包含该元素的B集合索引的映射，把「遍历所有B集合检查交集」的逻辑，转换为「通过A集合的元素直接查找对应B集合」，时间复杂度降至 O(MK + NL)（L为B集合的平均元素数），性能会有质的飞跃。

具体实现步骤 #

1. 预构建反向映射：遍历B中的每个集合，将集合内的每个元素作为键，对应的B集合索引作为值存入字典（存索引而非集合本身，可大幅节省内存）。
2. 批量查询匹配集合：对A中的每个集合，收集其所有元素对应的B集合索引，去重后再从B中取出对应集合。

代码示例 #

import secrets
from collections import defaultdict

# 生成样本数据函数
def generate_random_strings(num_strings, string_length):
    return set(secrets.token_hex(string_length) for _ in range(num_strings))

# 生成大规模测试数据
A = [generate_random_strings(5, 1) for _ in range(500000)]
B = [generate_random_strings(5, 1) for _ in range(500000)]

# 步骤1：构建元素到B集合索引的反向映射
element_to_b_indices = defaultdict(list)
for idx, b_set in enumerate(B):
    for elem in b_set:
        element_to_b_indices[elem].append(idx)

# 步骤2：处理A中的每个集合，获取匹配的B集合
def find_matching_sets(a_set):
    # 收集所有关联的B索引并去重
    matching_indices = set()
    for elem in a_set:
        if elem in element_to_b_indices:
            matching_indices.update(element_to_b_indices[elem])
    # 根据索引返回对应B集合
    return [B[idx] for idx in matching_indices]

# 单进程版本
results = [find_matching_sets(a) for a in A]

# 多进程版本（复用预构建的映射，提升效率）
from tqdm.contrib.concurrent import process_map
results = process_map(find_matching_sets, A, chunksize=1000)

额外优化点 #

• 内存优化：存储B集合的索引而非集合本身，可大幅减少反向映射的内存占用，尤其当单个元素对应多个B集合时效果更明显。
• 多进程共享内存：使用multiprocessing.Manager或shared_memory共享反向映射，避免每个进程拷贝大字典，节省内存并加快进程启动速度。
• 去重效率：用set对B集合索引去重，比列表遍历去重效率高得多，适合大规模数据场景。
• 哈希有效性：确保集合中的元素是可哈希类型（原代码中的字符串符合要求），若使用自定义类型需实现__hash__和__eq__方法。

内容的提问来源于stack exchange，提问作者do-me