得物Java面试被问：并查集的路径压缩和按秩合并

一、并查集基础与核心操作

1.1 并查集基本概念

python

复制

下载

class UnionFindBasic:
"""并查集基础实现"""

def __init__(self, n: int):
"""
初始化并查集
Args:
n: 元素数量，元素编号从0到n-1
"""
self.parent = list(range(n)) # 父节点数组
self.rank = [0] * n # 秩（树的高度）
self.count = n # 连通分量数量

def find_naive(self, x: int) -> int:
"""
朴素查找（未优化）
时间复杂度: O(h)，h为树高度，最坏O(n)
"""
while self.parent[x] != x:
x = self.parent[x]
return x

def union_naive(self, x: int, y: int) -> bool:
"""
朴素合并（未优化）
时间复杂度: O(h)
Returns:
bool: 如果x和y原本就在同一集合，返回False，否则返回True
"""
root_x = self.find_naive(x)
root_y = self.find_naive(y)

if root_x == root_y:
return False # 已经在同一集合

self.parent[root_x] = root_y
self.count -= 1
return True

def problems_with_naive(self):
"""朴素实现的问题"""
problems = {
"查找效率低": {
"原因": "树可能退化成链表",
"示例": "连续union(0,1), union(0,2), union(0,3)…",
"结果": "find操作需要遍历整个链表，O(n)"
},
"合并效率低": {
"原因": "总是将第一个集合合并到第二个",
"问题": "可能产生很高的树",
"影响": "后续find操作变慢"
},
"总体性能": {
"m次操作": "最坏O(mn)",
"无法处理大规模数据": "m=10^6, n=10^5时无法接受"
}
}
return problems

二、路径压缩优化

2.1 路径压缩原理

python

复制

下载

class PathCompression:
"""路径压缩优化"""

def __init__(self):
self.compression_techniques = self._explain_compression()

def _explain_compression(self):
"""解释路径压缩"""
return {
"核心思想": {
"目标": "使树变得更扁平",
"方法": "在find操作中，将路径上的所有节点直接指向根节点",
"效果": "下次查找这些节点时就是O(1)"
},

"数学原理": {
"摊还分析": {
"结论": "m次操作的摊还时间复杂度为O(mα(n))",
"α(n)": "反阿克曼函数，增长极慢",
"实际意义": "对于所有实际n，α(n) ≤ 5"
},
"复杂度证明": {
"均摊分析": "使用势能法或记账法",
"关键引理": "秩的性质和路径压缩的关系",
"最终结果": "O(mα(n))，几乎是线性的"
}
},

"压缩时机": {
"查找时压缩": "最常见的实现方式",
"合并时压缩": "可以顺便压缩",
"懒惰压缩": "需要时再压缩"
}
}

def find_recursive(self, parent: list, x: int) -> int:
"""
递归实现的路径压缩
优点：代码简洁
缺点：可能栈溢出（Python默认递归深度~1000）
"""
if parent[x] != x:
# 递归查找并压缩路径
parent[x] = self.find_recursive(parent, parent[x])
return parent[x]

def find_iterative(self, parent: list, x: int) -> int:
"""
迭代实现的路径压缩
优点：无递归深度限制
缺点：代码稍复杂
"""
# 第一步：找到根节点
root = x
while parent[root] != root:
root = parent[root]

# 第二步：路径压缩
while parent[x] != root:
parent_x = parent[x]
parent[x] = root
x = parent_x

return root

def find_two_pass(self, parent: list, x: int) -> int:
"""
两趟遍历的路径压缩
更容易理解和实现
"""
# 第一趟：找到根节点
root = x
while parent[root] != root:
root = parent[root]

# 第二趟：压缩路径
while parent[x] != root:
next_node = parent[x]
parent[x] = root
x = next_node

return root

def visualize_compression(self, parent: list, x: int):
"""可视化路径压缩过程"""
steps = []
original_path = []

# 记录原始路径
current = x
while parent[current] != current:
original_path.append(current)
current = parent[current]
root = current
original_path.append(root)

steps.append(f"原始路径: {' -> '.join(map(str, original_path))}")
steps.append(f"树高度: {len(original_path)-1}")

# 执行压缩
self.find_iterative(parent, x)

# 记录压缩后路径
compressed_path = [x]
if parent[x] != x:
compressed_path.append(parent[x])

steps.append(f"压缩后路径: {' -> '.join(map(str, compressed_path))}")
steps.append(f"新树高度: {len(compressed_path)-1}")

return steps

2.2 路径压缩实现

python

复制

下载

class UnionFindWithPathCompression:
"""带路径压缩的并查集"""

def __init__(self, n: int):
self.parent = list(range(n))
self.count = n

def find(self, x: int) -> int:
"""
带路径压缩的查找
使用递归实现（简洁）
"""
if self.parent[x] != x:
self.parent[x] = self.find(self.parent[x])
return self.parent[x]

def find_iterative_compression(self, x: int) -> int:
"""
带路径压缩的查找（迭代版）
避免递归深度限制
"""
# 找到根节点
root = x
while self.parent[root] != root:
root = self.parent[root]

# 压缩从x到根的路径
while self.parent[x] != root:
parent_x = self.parent[x]
self.parent[x] = root
x = parent_x

return root

def union(self, x: int, y: int) -> bool:
"""合并操作（使用路径压缩）"""
root_x = self.find(x)
root_y = self.find(y)

if root_x == root_y:
return False

# 简单合并：总是把x的根接到y的根
self.parent[root_x] = root_y
self.count -= 1
return True

def performance_analysis(self, n: int, operations: list):
"""性能分析"""
import time

start = time.time()
for op in operations:
if op[0] == 'find':
self.find(op[1])
elif op[0] == 'union':
self.union(op[1], op[2])

elapsed = time.time() – start

# 计算平均树高度
heights = []
for i in range(n):
height = 0
x = i
while self.parent[x] != x:
x = self.parent[x]
height += 1
heights.append(height)

avg_height = sum(heights) / n
max_height = max(heights)

return {
'total_time': elapsed,
'avg_tree_height': avg_height,
'max_tree_height': max_height,
'operations_count': len(operations)
}

篇幅限制下面就只能给大家展示小册部分内容了。整理了一份核心面试笔记包括了：Java面试、Spring、JVM、MyBatis、Redis、MySQL、并发编程、微服务、Linux、Springboot、SpringCloud、MQ、Kafc

需要全套面试笔记及答案 【点击此处即可/免费获取】​​​

三、按秩合并优化

3.1 按秩合并原理

python

复制

下载

class UnionByRank:
"""按秩合并优化"""

def __init__(self):
self.rank_explanation = self._explain_rank()

def _explain_rank(self):
"""解释按秩合并"""
return {
"秩的定义": {
"树的高度": "从叶子到根的最长路径长度",
"节点数量": "有时也使用节点数量作为秩",
"意义": "衡量树的"大小"或"高度""
},

"合并策略": {
"原则": "总是将矮树合并到高树下",
"目的": "避免树的高度增加",
"规则": [
"如果rank[root_x] < rank[root_y]: parent[root_x] = root_y",
"如果rank[root_x] > rank[root_y]: parent[root_y] = root_x",
"如果rank[root_x] == rank[root_y]: 任选一个作为根，rank加1"
]
},

"数学性质": {
"引理1": "秩为r的树至少包含2^r个节点",
"证明": "归纳法，合并时秩相同才增加",
"推论": "树的高度为O(log n)",

"引理2": "有n个节点的树，最大秩为floor(log₂n)",
"证明": "由引理1直接推出",
"意义": "即使没有路径压缩，树高也是对数级"
},

"与路径压缩的关系": {
"独立优化": "可以单独使用任一种优化",
"组合使用": "同时使用效果最佳",
"摊还复杂度": "O(α(n))，几乎是常数"
}
}

def union_by_height(self, parent: list, height: list, x: int, y: int):
"""按高度合并"""
root_x = self._find(parent, x)
root_y = self._find(parent, y)

if root_x == root_y:
return False

if height[root_x] < height[root_y]:
parent[root_x] = root_y
elif height[root_x] > height[root_y]:
parent[root_y] = root_x
else:
# 高度相等，任选一个作为根
parent[root_y] = root_x
height[root_x] += 1

return True

def union_by_size(self, parent: list, size: list, x: int, y: int):
"""按大小合并（节点数量）"""
root_x = self._find(parent, x)
root_y = self._find(parent, y)

if root_x == root_y:
return False

# 将小树合并到大树下
if size[root_x] < size[root_y]:
parent[root_x] = root_y
size[root_y] += size[root_x]
else:
parent[root_y] = root_x
size[root_x] += size[root_y]

return True

def _find(self, parent: list, x: int) -> int:
"""辅助查找函数"""
while parent[x] != x:
x = parent[x]
return x

def compare_strategies(self, n: int = 10000):
"""比较不同合并策略"""
import random

strategies = {
'按高度合并': self.union_by_height,
'按大小合并': self.union_by_size
}

results = {}

for name, strategy in strategies.items():
parent = list(range(n))

if '高度' in name:
rank = [0] * n
else:
rank = [1] * n # 初始每个节点大小为1

operations = 0
max_height = 0

# 执行随机合并操作
for _ in range(n // 2):
x = random.randint(0, n-1)
y = random.randint(0, n-1)

if strategy(parent, rank, x, y):
operations += 1

# 统计最大树高
if '高度' in name:
current_max = max(rank)
else:
# 对于按大小合并，需要计算实际高度
heights = []
for i in range(n):
h = 0
p = i
while parent[p] != p:
p = parent[p]
h += 1
heights.append(h)
current_max = max(heights)

max_height = max(max_height, current_max)

results[name] = {
'operations': operations,
'max_tree_height': max_height,
'theoretical_bound': 'O(log n)'
}

return results

3.2 按秩合并实现

python

复制

下载

class UnionFindWithRank:
"""带按秩合并的并查集"""

def __init__(self, n: int, use_size: bool = False):
"""
Args:
n: 元素数量
use_size: True表示按大小合并，False表示按高度合并
"""
self.parent = list(range(n))
self.use_size = use_size

if use_size:
self.rank = [1] * n # 存储子树大小
else:
self.rank = [0] * n # 存储树的高度

self.count = n

def find(self, x: int) -> int:
"""查找（不带路径压缩，用于对比）"""
while self.parent[x] != x:
x = self.parent[x]
return x

def union(self, x: int, y: int) -> bool:
"""按秩合并"""
root_x = self.find(x)
root_y = self.find(y)

if root_x == root_y:
return False

if self.use_size:
# 按大小合并
if self.rank[root_x] < self.rank[root_y]:
self.parent[root_x] = root_y
self.rank[root_y] += self.rank[root_x]
else:
self.parent[root_y] = root_x
self.rank[root_x] += self.rank[root_y]
else:
# 按高度合并
if self.rank[root_x] < self.rank[root_y]:
self.parent[root_x] = root_y
elif self.rank[root_x] > self.rank[root_y]:
self.parent[root_y] = root_x
else:
# 高度相等，任选一个作为根，高度加1
self.parent[root_y] = root_x
self.rank[root_x] += 1

self.count -= 1
return True

def get_tree_stats(self):
"""获取树统计信息"""
n = len(self.parent)
heights = []
sizes = []

# 找到所有根节点
roots = set()
for i in range(n):
root = self.find(i)
roots.add(root)

# 计算每棵树的高度和大小
for root in roots:
# 计算树高度（BFS）
from collections import deque

height = 0
size = 0

# 找到树中所有节点
nodes = [i for i in range(n) if self.find(i) == root]
size = len(nodes)

if size == 1:
heights.append(0)
sizes.append(1)
continue

# 构建树的邻接表
tree = {i: [] for i in nodes}
for i in nodes:
if self.parent[i] != i:
tree[self.parent[i]].append(i)
tree[i].append(self.parent[i])

# BFS求高度
visited = set()
queue = deque([(root, 0)])
max_depth = 0

while queue:
node, depth = queue.popleft()
if node in visited:
continue
visited.add(node)
max_depth = max(max_depth, depth)

for neighbor in tree[node]:
if neighbor not in visited:
queue.append((neighbor, depth + 1))

heights.append(max_depth)
sizes.append(size)

return {
'total_trees': len(roots),
'avg_height': sum(heights) / len(heights) if heights else 0,
'max_height': max(heights) if heights else 0,
'avg_size': sum(sizes) / len(sizes) if sizes else 0,
'max_size': max(sizes) if sizes else 0,
'min_size': min(sizes) if sizes else 0
}

四、双重优化：路径压缩 + 按秩合并

4.1 完整优化实现

python

复制

下载

class OptimizedUnionFind:
"""双重优化的并查集：路径压缩 + 按秩合并"""

def __init__(self, n: int, union_by: str = 'rank'):
"""
Args:
n: 元素数量
union_by: 'rank'按高度合并，'size'按大小合并
"""
self.parent = list(range(n))
self.union_by = union_by

if union_by == 'size':
self.rank = [1] * n # 存储子树大小
else: # 'rank'
self.rank = [0] * n # 存储树的高度

self.count = n
self.operations = 0 # 操作计数器

def find(self, x: int) -> int:
"""
查找 + 路径压缩
时间复杂度: 摊还O(α(n))
"""
self.operations += 1

# 路径压缩（递归实现）
if self.parent[x] != x:
self.parent[x] = self.find(self.parent[x])
return self.parent[x]

def find_iterative(self, x: int) -> int:
"""
查找 + 路径压缩（迭代实现）
避免递归深度限制
"""
self.operations += 1

# 找到根节点
root = x
while self.parent[root] != root:
root = self.parent[root]

# 压缩路径
while self.parent[x] != root:
parent_x = self.parent[x]
self.parent[x] = root
x = parent_x

return root

def union(self, x: int, y: int) -> bool:
"""
合并 + 按秩合并
时间复杂度: O(α(n))
"""
self.operations += 1

root_x = self.find(x)
root_y = self.find(y)

if root_x == root_y:
return False

if self.union_by == 'size':
# 按大小合并
if self.rank[root_x] < self.rank[root_y]:
self.parent[root_x] = root_y
self.rank[root_y] += self.rank[root_x]
else:
self.parent[root_y] = root_x
self.rank[root_x] += self.rank[root_y]
else:
# 按高度合并
if self.rank[root_x] < self.rank[root_y]:
self.parent[root_x] = root_y
elif self.rank[root_x] > self.rank[root_y]:
self.parent[root_y] = root_x
else:
# 高度相等，任选一个作为根
self.parent[root_y] = root_x
self.rank[root_x] += 1

self.count -= 1
return True

def is_connected(self, x: int, y: int) -> bool:
"""判断两个元素是否连通"""
return self.find(x) == self.find(y)

def get_component_size(self, x: int) -> int:
"""获取包含x的连通分量大小"""
root = self.find(x)
if self.union_by == 'size':
return self.rank[root]
else:
# 需要遍历计算
count = 0
for i in range(len(self.parent)):
if self.find(i) == root:
count += 1
return count

def benchmark(self, operations: list):
"""性能基准测试"""
import time

start = time.perf_counter()

for op in operations:
if op[0] == 'find':
self.find(op[1])
elif op[0] == 'union':
self.union(op[1], op[2])
elif op[0] == 'connected':
self.is_connected(op[1], op[2])

elapsed = time.perf_counter() – start

# 分析树结构
stats = self._analyze_structure()

return {
'total_time': elapsed,
'operations_per_second': len(operations) / elapsed if elapsed > 0 else float('inf'),
'total_operations': self.operations,
'components': self.count,
'structure_stats': stats
}

def _analyze_structure(self):
"""分析树结构"""
n = len(self.parent)

# 计算平均查找路径长度
total_path_length = 0
max_path_length = 0

for i in range(n):
path_len = 0
x = i
while self.parent[x] != x:
x = self.parent[x]
path_len += 1
total_path_length += path_len
max_path_length = max(max_path_length, path_len)

avg_path_length = total_path_length / n

# 计算秩的分布
rank_distribution = {}
for r in self.rank:
rank_distribution[r] = rank_distribution.get(r, 0) + 1

return {
'avg_path_length': avg_path_length,
'max_path_length': max_path_length,
'avg_rank': sum(self.rank) / n,
'max_rank': max(self.rank),
'rank_distribution': dict(sorted(rank_distribution.items()))
}

 篇幅限制下面就只能给大家展示小册部分内容了。整理了一份核心面试笔记包括了：Java面试、Spring、JVM、MyBatis、Redis、MySQL、并发编程、微服务、Linux、Springboot、SpringCloud、MQ、Kafc

需要全套面试笔记及答案 【点击此处即可/免费获取】​​​

4.2 复杂度分析与证明

python

复制

下载

class UnionFindComplexity:
"""并查集复杂度分析"""

def __init__(self):
self.complexity_analysis = self._analyze_complexity()

def _analyze_complexity(self):
"""复杂度分析"""
return {
"朴素实现": {
"find": "O(h)，最坏O(n)",
"union": "O(h)，最坏O(n)",
"m次操作": "O(mn)，最坏情况"
},

"仅按秩合并": {
"find": "O(log n)，树高≤log₂n",
"union": "O(log n)",
"m次操作": "O(m log n)",
"证明": "引理：秩为r的树至少包含2^r个节点"
},

"仅路径压缩": {
"find": "摊还O(α(n))",
"union": "摊还O(α(n))",
"m次操作": "O(mα(n))",
"注意": "单独路径压缩的证明非常复杂"
},

"双重优化": {
"find": "摊还O(α(n))",
"union": "摊还O(α(n))",
"m次操作": "O(mα(n))",
"α(n)": "反阿克曼函数，对于实际n值≤5",
"实际意义": "基本上是常数时间"
},

"反阿克曼函数": {
"定义": """
α(n) = min{k ≥ 0: A(k, ⌊n⌋) > log₂n}
其中A是阿克曼函数
""",
"增长": "增长极慢",
"数值": """
n = 1: α(n) = 0
n = 3: α(n) = 1
n = 7: α(n) = 2
n = 2047: α(n) = 3
n = 2^65536: α(n) = 4
对于所有实际n：α(n) ≤ 5
"""
}
}

def amortized_analysis_proof(self):
"""摊还分析证明概要"""
proof_steps = [
"定理：使用路径压缩和按秩合并的并查集，",
"m次操作的摊还时间复杂度为O(mα(n))",
"",
"证明概要（使用势能法）：",
"",
"1. 定义势函数：",
" Φ = Σ φ(x)，其中φ(x)依赖于x的秩和父节点的秩",
"",
"2. 定义秩的性质：",
" (a) 0 ≤ rank[x] ≤ ⌊log₂n⌋",
" (b) rank[parent[x]] ≥ rank[x]",
" (c) 有特定秩的节点数有限",
"",
"3. 分析操作代价：",
" – find操作：实际代价O(路径长度)",
" – 势能变化：路径压缩降低势能",
" – 摊还代价：O(α(n))",
"",
"4. union操作：",
" – 实际代价：两次find + 常数",
" – 摊还代价：O(α(n))",
"",
"5. 总摊还代价：",
" O(mα(n))"
]

return "\\n".join(proof_steps)

def empirical_validation(self, max_n: int = 10**6):
"""经验验证"""
import random
import time

results = []

for n in [100, 1000, 10000, 100000, 1000000]:
if n > max_n:
break

# 创建不同实现的并查集
implementations = {
'朴素': UnionFindBasic(n),
'仅路径压缩': UnionFindWithPathCompression(n),
'仅按秩合并': UnionFindWithRank(n),
'双重优化': OptimizedUnionFind(n)
}

# 生成随机操作序列
operations = []
for _ in range(n * 2):
op_type = random.choice(['find', 'union'])
if op_type == 'find':
x = random.randint(0, n-1)
operations.append(('find', x))
else:
x = random.randint(0, n-1)
y = random.randint(0, n-1)
operations.append(('union', x, y))

# 测试每个实现
impl_results = {}
for name, uf in implementations.items():
start = time.perf_counter()

for op in operations:
if op[0] == 'find':
if hasattr(uf, 'find_naive'):
uf.find_naive(op[1])
else:
uf.find(op[1])
else: # union
if hasattr(uf, 'union_naive'):
uf.union_naive(op[1], op[2])
else:
uf.union(op[1], op[2])

elapsed = time.perf_counter() – start
impl_results[name] = elapsed

results.append({
'n': n,
'operations': len(operations),
'times': impl_results,
'speedup': {
'双重优化/朴素': impl_results['朴素'] / impl_results['双重优化'],
'双重优化/仅秩合并': impl_results['仅按秩合并'] / impl_results['双重优化'],
'双重优化/仅路径压缩': impl_results['仅路径压缩'] / impl_results['双重优化']
}
})

return results

五、实际应用与变种

5.1 经典应用场景

python

复制

下载

class UnionFindApplications:
"""并查集应用场景"""

def kruskal_mst(self, n: int, edges: list):
"""
Kruskal最小生成树算法
时间复杂度: O(E log E + E α(V))
"""
# 按权重排序边
edges.sort(key=lambda x: x[2])

uf = OptimizedUnionFind(n)
mst = []
mst_weight = 0

for u, v, weight in edges:
if uf.union(u, v):
mst.append((u, v, weight))
mst_weight += weight

if len(mst) == n – 1:
break

return {
'mst_edges': mst,
'total_weight': mst_weight,
'components': uf.count
}

def connected_components(self, n: int, edges: list):
"""计算连通分量"""
uf = OptimizedUnionFind(n)

for u, v in edges:
uf.union(u, v)

# 统计每个连通分量
components = {}
for i in range(n):
root = uf.find(i)
if root not in components:
components[root] = []
components[root].append(i)

return {
'component_count': uf.count,
'components': list(components.values()),
'largest_component': max(len(comp) for comp in components.values()),
'component_sizes': [len(comp) for comp in components.values()]
}

def detect_cycle(self, n: int, edges: list):
"""检测无向图中的环"""
uf = OptimizedUnionFind(n)

for u, v in edges:
if not uf.union(u, v):
return {
'has_cycle': True,
'cycle_edge': (u, v),
'components': uf.count
}

return {
'has_cycle': False,
'components': uf.count
}

def percolation_problem(self, n: int, open_sites: list):
"""
渗流问题
判断n×n网格是否从上到下连通
"""
# 创建n×n+2的并查集（+2表示虚拟的顶部和底部）
uf = OptimizedUnionFind(n * n + 2)
top = n * n
bottom = n * n + 1

# 连接顶部行和底部行
for col in range(n):
uf.union(top, col) # 第一行
uf.union(bottom, (n-1)*n + col) # 最后一行

# 打开站点并连接相邻的打开站点
grid = [[False] * n for _ in range(n)]

for row, col in open_sites:
grid[row][col] = True
index = row * n + col

# 连接上下左右相邻的打开站点
directions = [(-1, 0), (1, 0), (0, -1), (0, 1)]
for dr, dc in directions:
nr, nc = row + dr, col + dc
if 0 <= nr < n and 0 <= nc < n and grid[nr][nc]:
neighbor_index = nr * n + nc
uf.union(index, neighbor_index)

# 检查是否渗流
percolates = uf.is_connected(top, bottom)

return {
'percolates': percolates,
'open_sites_count': len(open_sites),
'open_fraction': len(open_sites) / (n * n)
}

5.2 扩展变种

python

复制

下载

class ExtendedUnionFind:
"""扩展的并查集变种"""

def __init__(self, n: int):
self.parent = list(range(n))
self.rank = [0] * n

# 扩展属性
self.size = [1] * n # 每个集合的大小
self.min = list(range(n)) # 每个集合的最小元素
self.max = list(range(n)) # 每个集合的最大元素
self.sum = list(range(n)) # 每个集合的元素和（假设元素值等于索引）

def find(self, x: int) -> int:
"""带路径压缩的查找"""
if self.parent[x] != x:
self.parent[x] = self.find(self.parent[x])
return self.parent[x]

def union(self, x: int, y: int) -> bool:
"""按秩合并，并维护扩展属性"""
root_x = self.find(x)
root_y = self.find(y)

if root_x == root_y:
return False

# 按秩合并
if self.rank[root_x] < self.rank[root_y]:
root_x, root_y = root_y, root_x

self.parent[root_y] = root_x

if self.rank[root_x] == self.rank[root_y]:
self.rank[root_x] += 1

# 维护扩展属性
self.size[root_x] += self.size[root_y]
self.min[root_x] = min(self.min[root_x], self.min[root_y])
self.max[root_x] = max(self.max[root_x], self.max[root_y])
self.sum[root_x] += self.sum[root_y]

return True

def get_component_info(self, x: int) -> dict:
"""获取包含x的连通分量的所有信息"""
root = self.find(x)

return {
'root': root,
'size': self.size[root],
'min_element': self.min[root],
'max_element': self.max[root],
'sum': self.sum[root],
'avg': self.sum[root] / self.size[root] if self.size[root] > 0 else 0
}

def dynamic_connectivity(self, n: int, operations: list):
"""
动态连通性问题
支持查询和边添加，但不支持删除
"""
results = []

for op in operations:
if op[0] == 'add':
u, v = op[1], op[2]
self.union(u, v)
results.append(f"Added edge ({u}, {v})")
elif op[0] == 'query':
u, v = op[1], op[2]
connected = self.find(u) == self.find(v)
results.append(f"Query ({u}, {v}): {connected}")
elif op[0] == 'component':
x = op[1]
info = self.get_component_info(x)
results.append(f"Component of {x}: {info}")

return results

class WeightedUnionFind:
"""带权并查集（维护相对关系）"""

def __init__(self, n: int):
self.parent = list(range(n))
self.rank = [0] * n
self.weight = [0] * n # weight[x] 表示 x 到 parent[x] 的权值

def find(self, x: int) -> int:
"""带权值的路径压缩"""
if self.parent[x] != x:
root = self.find(self.parent[x])
self.weight[x] += self.weight[self.parent[x]]
self.parent[x] = root
return self.parent[x]

def union(self, x: int, y: int, w: float) -> bool:
"""
合并x和y，已知x和y之间的权值关系为: value[x] + w = value[y]
"""
root_x = self.find(x)
root_y = self.find(y)

if root_x == root_y:
# 检查一致性
return abs((self.weight[x] – self.weight[y]) – w) < 1e-9

# 按秩合并
if self.rank[root_x] < self.rank[root_y]:
root_x, root_y = root_y, root_x
x, y = y, x
w = -w

self.parent[root_y] = root_x

# 更新权值：需要满足 weight[x] + w = weight[y] + new_weight
# 所以 new_weight = weight[x] – weight[y] + w
self.weight[root_y] = self.weight[x] – self.weight[y] + w

if self.rank[root_x] == self.rank[root_y]:
self.rank[root_x] += 1

return True

def diff(self, x: int, y: int) -> float:
"""
返回 value[y] – value[x]
如果x和y不在同一集合，返回None
"""
if self.find(x) != self.find(y):
return None
return self.weight[y] – self.weight[x]

六、面试实战

6.1 常见面试题

python

复制

下载

class UnionFindInterviewProblems:
"""并查集面试题目"""

def number_of_islands(self, grid: list) -> int:
"""
LeetCode 200: 岛屿数量
使用并查集解决
"""
if not grid:
return 0

m, n = len(grid), len(grid[0])

# 初始化并查集
uf = OptimizedUnionFind(m * n)

# 方向：右，下
directions = [(0, 1), (1, 0)]

# 统计陆地数量
land_count = 0

for i in range(m):
for j in range(n):
if grid[i][j] == '1':
land_count += 1
index = i * n + j

# 连接相邻的陆地
for di, dj in directions:
ni, nj = i + di, j + dj
if 0 <= ni < m and 0 <= nj < n and grid[ni][nj] == '1':
neighbor_index = ni * n + nj
if uf.union(index, neighbor_index):
# 每成功合并一次，陆地数量减1
land_count -= 1

return land_count

def longest_consecutive_sequence(self, nums: list) -> int:
"""
LeetCode 128: 最长连续序列
使用并查集连接相邻的数字
"""
if not nums:
return 0

# 创建数字到索引的映射
num_to_index = {}
uf = OptimizedUnionFind(len(nums))

# 第一遍：建立映射
for i, num in enumerate(nums):
if num not in num_to_index:
num_to_index[num] = i

# 第二遍：连接相邻数字
for i, num in enumerate(nums):
# 连接num-1
if num – 1 in num_to_index:
uf.union(i, num_to_index[num – 1])

# 连接num+1
if num + 1 in num_to_index:
uf.union(i, num_to_index[num + 1])

# 统计每个连通分量的大小
from collections import defaultdict
component_sizes = defaultdict(int)

for i in range(len(nums)):
root = uf.find(i)
component_sizes[root] += 1

return max(component_sizes.values()) if component_sizes else 0

def accounts_merge(self, accounts: list) -> list:
"""
LeetCode 721: 账户合并
使用并查集合并有相同邮箱的账户
"""
# 邮箱 -> 账户索引的映射
email_to_index = {}
n = len(accounts)
uf = OptimizedUnionFind(n)

# 第一遍：构建并查集
for i, account in enumerate(accounts):
name, emails = account[0], account[1:]
for email in emails:
if email in email_to_index:
# 这个邮箱已经出现过，合并账户
uf.union(i, email_to_index[email])
else:
email_to_index[email] = i

# 第二遍：收集每个账户的所有邮箱
from collections import defaultdict
index_to_emails = defaultdict(set)

for email, idx in email_to_index.items():
root = uf.find(idx)
index_to_emails[root].add(email)

# 第三遍：构造结果
result = []
for idx, emails in index_to_emails.items():
name = accounts[idx][0]
result.append([name] + sorted(list(emails)))

return result

def friend_requests(self, n: int, restrictions: list, requests: list) -> list:
"""
LeetCode 2076: 处理好友请求
使用带权并查集（敌人关系）
"""
uf = OptimizedUnionFind(n)

# 构建初始的敌人关系
enemy = [-1] * n # enemy[i] 表示i的敌人集合的代表

for a, b in restrictions:
# 找到a和b的根
root_a = uf.find(a)
root_b = uf.find(b)

# 设置敌人关系
if enemy[root_a] == -1:
enemy[root_a] = root_b
else:
uf.union(enemy[root_a], root_b)

if enemy[root_b] == -1:
enemy[root_b] = root_a
else:
uf.union(enemy[root_b], root_a)

# 处理请求
results = []

for u, v in requests:
root_u = uf.find(u)
root_v = uf.find(v)

if root_u == root_v:
# 已经在同一朋友圈
results.append(True)
elif enemy[root_u] == root_v or enemy[root_v] == root_u:
# 是敌人关系
results.append(False)
else:
# 可以成为朋友，合并朋友圈
uf.union(root_u, root_v)

# 合并敌人关系
new_root = uf.find(root_u)
other_root = root_v if new_root == root_u else root_u

if enemy[root_u] != -1 and enemy[root_v] != -1:
uf.union(enemy[root_u], enemy[root_v])
enemy[new_root] = uf.find(enemy[root_u])
enemy[enemy[new_root]] = new_root
elif enemy[root_u] != -1:
enemy[new_root] = enemy[root_u]
enemy[enemy[new_root]] = new_root
elif enemy[root_v] != -1:
enemy[new_root] = enemy[root_v]
enemy[enemy[new_root]] = new_root

results.append(True)

return results

6.2 面试回答模板

python

复制

下载

class UnionFindInterviewGuide:
"""并查集面试指南"""

@staticmethod
def explain_optimizations():
"""解释优化"""
template = """
并查集有两种核心优化：路径压缩和按秩合并

1. 路径压缩 (Path Compression):
– 在find操作中，将路径上的所有节点直接指向根节点
– 使树变得扁平，后续查找更快
– 实现方式：
* 递归：if parent[x] != x: parent[x] = find(parent[x])
* 迭代：两趟遍历，先找根，再压缩

2. 按秩合并 (Union by Rank):
– 在union操作中，总是将矮树合并到高树下
– 避免树的高度不必要增加
– "秩"可以是树高度或节点数量
– 实现规则：
* 如果rank[x] < rank[y]: parent[x] = y
* 如果rank[x] > rank[y]: parent[y] = x
* 如果相等：任选一个，rank加1

3. 双重优化的复杂度：
– m次操作的摊还时间复杂度：O(mα(n))
– α(n)是反阿克曼函数，增长极慢
– 对于实际n值，α(n) ≤ 5，基本上是常数

4. 为什么要两种都用：
– 单独路径压缩：最坏情况可能仍需要O(log n)
– 单独按秩合并：保证树高O(log n)
– 两者结合：达到最优的摊还复杂度
"""
return template

@staticmethod
def coding_implementation():
"""编码实现"""
code = """
class UnionFind:
def __init__(self, n):
self.parent = list(range(n))
self.rank = [0] * n # 按高度合并
# 或者 self.size = [1] * n # 按大小合并

def find(self, x):
# 路径压缩（递归）
if self.parent[x] != x:
self.parent[x] = self.find(self.parent[x])
return self.parent[x]

def find_iterative(self, x):
# 路径压缩（迭代）
root = x
while self.parent[root] != root:
root = self.parent[root]

# 压缩路径
while self.parent[x] != root:
parent_x = self.parent[x]
self.parent[x] = root
x = parent_x

return root

def union(self, x, y):
root_x = self.find(x)
root_y = self.find(y)

if root_x == root_y:
return False

# 按秩合并（高度）
if self.rank[root_x] < self.rank[root_y]:
self.parent[root_x] = root_y
elif self.rank[root_x] > self.rank[root_y]:
self.parent[root_y] = root_x
else:
self.parent[root_y] = root_x
self.rank[root_x] += 1

return True
"""
return code

@staticmethod
def common_questions():
"""常见问题"""
return {
"基础问题": [
"并查集的时间复杂度是多少？如何证明？",
"路径压缩和按秩合并分别解决了什么问题？",
"为什么需要两种优化都使用？",
"反阿克曼函数是什么？为什么重要？"
],

"实现问题": [
"如何实现递归和非递归的路径压缩？",
"按高度合并和按大小合并有什么区别？",
"如何处理大规模数据（n>10^7）？",
"并查集的空间复杂度是多少？"
],

"应用问题": [
"如何用并查集解决岛屿数量问题？",
"如何用并查集检测图中是否有环？",
"带权并查集有什么应用场景？",
"如何用并查集维护集合的统计信息？"
],

"扩展问题": [
"如何支持删除操作？",
"如何实现持久化并查集？",
"并行环境下的并查集如何设计？",
"并查集在分布式系统中的应用？"
]
}

 篇幅限制下面就只能给大家展示小册部分内容了。整理了一份核心面试笔记包括了：Java面试、Spring、JVM、MyBatis、Redis、MySQL、并发编程、微服务、Linux、Springboot、SpringCloud、MQ、Kafc

需要全套面试笔记及答案 【点击此处即可/免费获取】​​​

七、性能对比与总结

7.1 性能对比实验

python

复制

下载

class UnionFindBenchmark:
"""并查集性能对比"""

@staticmethod
def run_comparison():
"""运行性能对比"""
import random
import time

n = 1000000
operations = []

# 生成随机操作序列
for _ in range(2000000):
op_type = random.choice(['find', 'union'])
if op_type == 'find':
x = random.randint(0, n-1)
operations.append(('find', x))
else:
x = random.randint(0, n-1)
y = random.randint(0, n-1)
operations.append(('union', x, y))

implementations = {
'Naive': UnionFindBasic(n),
'Path Compression Only': UnionFindWithPathCompression(n),
'Union by Rank Only': UnionFindWithRank(n),
'Both Optimizations': OptimizedUnionFind(n)
}

results = {}

for name, uf in implementations.items():
print(f"Testing {name}…")

start = time.perf_counter()

for op in operations:
if op[0] == 'find':
if hasattr(uf, 'find_naive'):
uf.find_naive(op[1])
else:
uf.find(op[1])
else:
if hasattr(uf, 'union_naive'):
uf.union_naive(op[1], op[2])
else:
uf.union(op[1], op[2])

elapsed = time.perf_counter() – start

# 统计树高度
max_height = 0
if hasattr(uf, 'parent'):
for i in range(min(1000, n)): # 抽样检查
height = 0
x = i
while uf.parent[x] != x:
x = uf.parent[x]
height += 1
max_height = max(max_height, height)

results[name] = {
'time': elapsed,
'max_height': max_height,
'speed': len(operations) / elapsed
}

# 打印结果
print("\\n" + "="*60)
print("Performance Comparison Results:")
print("="*60)

fastest = min(results.items(), key=lambda x: x[1]['time'])

for name, data in results.items():
speedup = fastest[1]['time'] / data['time'] if data['time'] > 0 else 0
print(f"\\n{name}:")
print(f" Time: {data['time']:.3f}s")
print(f" Speed: {data['speed']:.0f} ops/s")
print(f" Max Height: {data['max_height']}")
print(f" Speedup vs Fastest: {speedup:.2f}x")

return results

这个并查集的路径压缩和按秩合并详解涵盖了从基础原理到高级优化的完整内容，特别适合面试准备和算法学习。关键要点总结：