21-《进阶》哈希、位图、布隆及岛.md

[TOC]

# 1 哈希函数

## 1.1 认识hash函数

1）输入参数data，假设是in类型，特征：可能性无穷大，比如str类型的参数

2）输出参数类型out，特征：可能性可以很大，但一定是有穷尽的

3）哈希函数没有任何随机的机制，固定的输入一定是固定的输出

4）输入无穷多但输出值有限，所以不同输入也可能输出相同（哈希碰撞）

5）再相似的不同输入，得到的输出值，会几乎均匀的分布在out域上（离散处理）

重点：第5条！所以hash函数又叫做散列函数

通过hash散列之后的值，如果模上一个数，模之后的数仍然是散列的！具有传递性


> 对于两个输入，即使长的再像，hash函数也会散列，均分分布在值域上


## 1.2 hash应用函数举例


假设有一个含有40亿个无符号整数的文件，每一个无符号整数含有4个字节，该文件每一行含有一个无符号整数。无符号整数的范围是0到42亿，即0到2的32次方减1的范围。假设我们有一个含有1G字节的内存，试着统计该文件中哪一个整数出现的次数最多


我们用这一个G内存，做一张hash表，key和value都是无符号整数。所以key和value都是4字节，每一条kv含有8字节，hash函数内部索引空间不计。假设我们文件中每一个无符号整数都不相同，需要40亿个kv，大约为8乘以40亿为320亿字节。转化为内存大约需要32G内存空间。所以我们不能直接通过1G内存直接统计，有可能内存会爆掉；


我们1G内存只能装大概1亿条8字节的kv。该文件大概需要32G内存，我们直接通过Hash函数散列后的值，模上40。范围变为0到39，编号0到39号的文件，每一个状态发送到一个文件。同一种数字key，肯定会发送到相同编号的文件中去。经过这样的处理，假设最差的情况文件中40亿个整数均不同，我们大概会发送到0到39的文件中，每个文件大概1亿个整数


用1G的内存，先统计0号子文件中的词频，再释放内存去统计1号，2号直到39号文件。每个文件最多不超过1亿条记录。1G内存不会爆掉。选出40个词频最高的数字，再进行处理


这样处理，即使文件中40亿个数都相同，落到相同的子文件0中去。那么1G内存统计相同的key，只会再原有基础上增大value，内存绝对不会爆掉


## 1.3 hash函数实现


> Hash函数使用的时候，理论上任务增删改查的时间复杂度都为O(1);


### 1.3.1 认识经典hash（数组加单链表）

假设我们的Hash的key的范围为17，我们新添加的key再这17个格子上，很大可能会出现碰撞，假设我们第一条需要保存的kv为`'abc':1`,我们用一个函数算出hash值，用该hash值模17，假设得到5。那么我们把这条记录放在5的位置；

同样的方式第二条kv`'def':32`，算出来的位置为10。第三条kv`'bbq':44`,算出的位置也为5，那么我们用单链表去串在之前`'abc':1`的下面


假设我们get我们的'abc',通过相同的函数算出hash值，再模17，肯定得到5，我们去5的下面去遍历链表，找到'abc'拿出它的值即可


由于使用的hash值进行填充，理论上我们17长度的数组，很容易就碰撞，而且每个格子均匀的碰撞。我们感知到某个位置上的链表长度大于等于6，可以认为其他格子的链表长度也差不多大于等于6。此时我们进行hash扩容，增大我们的数组长度。假设扩容为原数组的两倍，范围为0到34；

**紧接着我们用非常暴力的手段，把老的数组上的每个链表上的节点都取出来，重新计算hash值，模34，决定放在新数组的哪个位置，那么根据散列的性质，我们大概知道，样本分配到新的数组中，每个位置链表的长度，大概为3**


复杂度分析：对于一个key_value，算key的hash值认为O(1)；hash值模一个值O(1)；找到数组中的桶，而且桶长度不超过6，那么也是O(1)。所以不涉及到扩容，hash结增删改查操作严格O(1)；

**但是Hash函数是会涉及到扩容的，我们可以假设初始结构的数组长度为2进行推演，那么对于样本容量为N的hash结构，如果扩容，那么N长度的哈希结构会经历log2N,log以2为底的N次。每次扩容的代价时间复杂度为O(N), 对于N样本，之前所有扩容的总代价为O(N)✖log2N。均摊到每次hash值的增删改查，所有hash表考虑扩容的增删改查时间复杂度不是O(1)。而是`（O(N)✖log2N）➗ N`**


> 但是Hash表有很多改进，比如扩容倍数不是扩两倍，而是扩更多倍，用以减少扩容次数，从而减少log的底数。或者假设在Jvm中，用户申请的hash表不够用了，JVM离线扩容，用户无感知；或者我们在数组中放有序表而不是单链表，例如treeSet结构，我们长度就不需要到6再扩容了，等等。我们还是认为hash表在使用时，增删改查操作就是O(1)的，虽然理论上并不是

> 注意！！！上述估计hash扩容代价的订正，注意上述说扩容代价为O(logN)可以认为为O(1)，这种说法是错误的，hash扩容总代价是O(N)，均摊下来就是O(1)。因为当我们数据量已经扩容到N,之前的扩容是一次一次叠加而来，可以假设从样本为1的时候开始扩容，N的数据量对应的扩容就是1+2+...+n/4+n/2; 均摊到当前的N，就是O(1)

# 2 布隆过滤器

1）利用哈希函数的性质

2）每一条数据提取特征

3）加入描黑库

> 布隆过滤器解决什么问题？

类似没有删除的黑名单系统。常用来解决爬虫黑名单问题，例如我们的爬虫系统爬到一个url把它放到黑名单系统中，另外一个爬虫爬到该url时，先去检查有没有其他系统已经爬过。


假设我们有100亿个url构成的黑名单，首先考虑使用MapSet，一个url假设64字节，100亿个url,大小为6400亿字节，大约640GB内存空间。显然不太合适使用map结构。只有添加和查询，没有删除操作，我们构建布隆过滤器，大约该体量的查询只需要30G内存空间


## 2.1 位图的概念

假设我们申请int[100] arr的数组大小，每一个位置是int32位，100个位置总共需要3200bit；

假设我们想要知道arr的位图的453位是什么数字，那么我们可以立马知道453位bit的状态是0还是1，`int status = (arr[453/32] >> (453%32)) & 1;`；我们也可以设置453位的状态,比如我们要设置453位的状态为1：`arr[453/32] = arr[453/32] | (1 << (453%32));`;

对于数据量比较大的存储，我们可以使用二维数组，比如int[100][100] 表示的总共有320000个bit。我们想要把170039位置的比特拿出来，可以先求在哪一行，`170039/3200`等于53行，在53行的`（170039 % 3200 )/32 `位置，在该位置的`（170039 % 3200 )%32`bit的位置上


## 2.2 布隆过滤器

布隆过滤器建立在位图的概念上。

### 2.2.1 布隆过滤器的添加

假设我们有m长度的位图，初始化位图都为0，某一个位置需要置为1，就属于描黑的状态。我们给每一个url算出一个hash值，让该hash值模上m，决定一个位置进行描黑，再用另外一个hash函数算出hash值，模上m描黑第二个点，假设需要k个hash函数，那么一个url需要描黑k个位置，k个位置中有可能有重复的描黑点，在不同的hash函数算出相同的hash值的时候会出现这种情况。经过上述的处理，该url属于被加入我们的黑名单


### 2.2.2 布隆过滤器的查询

给定一个url，相同的我们用k个hash函数算出k个hash值，用这k个hash值模上m，算出k个位置。假设k个位置都属于描黑的状态，就任务该url已经添加进来我们的黑名单系统了

> 比如对于指纹来说，K个hash函数，可以类比为K个提取指纹的方式


布隆过滤器有可能出现误判，例如我们的m数组长度比较小，url比较多，那么所有url都加进来有可能m长度数组的位图全被描黑。那么一个新的url过来，我们误以为已经加入了进来。


> 布隆过滤器虽然有失误率，存在误判。但是不可能判断出一个url本身在黑名单，判断出来不在黑名单，只可能出现一个url不在黑名单，判断出在黑名单这种情况的误判，即宁可错杀三千，绝不放过一个

所以严格要求不予许有失误率的场景，用不了布隆过滤器


### 2.2.3 k个hash函数如何选择，位图m空间选择多大

我们可以根据样本量N，来设计我们的k和m的大小。设计布隆过滤器，我们必须提前知道样本量


预期失误率P


我们知道一个样本量大小为N，该布隆过滤器设计出来后，预期失误率不得大于P。那么我们把m当成横坐标，样本失误率P当成纵坐标。样本大小是常数N。我们可以感受到m越小P越大，M越大P越小，且P永远不可能为0。预期失误率p是纵坐标的一个常数点。**P和M该函数满足对数分布**


hash函数个数k和失误率的关系，P为纵坐标，k为横坐标。直观理解上，如果hash函数的个数K过小，由于提取特征不够过，失误率p会很大；如果k过多，那么需要描黑的点就过多，m会迅速耗尽。增大了失误率；**所以p和k的关系函数的图像是一个底朝上的二次函数分布**，注意是p和k，下面公式2，k是和m的关系函数


### 2.2.4 布隆过滤器重要公式

1，假设样本数据量为n，预期的失误率为p（布隆过滤器大小和每个样本的大小无关）

2，根据n和p，算出Bloom Filter一共需要多少个bit位，向上取整，记为m

3，根据m和n，算出Bloom Filter需要多少个哈希函数，向上取整，记为k

4，根据修正公式，算出真实的失误率p_true

公式1：

```math
m = -   \frac{n*lnp}{(ln2)^2}  
```

公式2：

```math
k = ln2 * \frac{m}{n}   = 0.7 * \frac{m}{n}
```

m和K算出来是一个小数，比如26.4G，和13.3。那么我们向上取整为27G或者30G。k向上取整为14个。注意m和k一定不能比理论值还要小了。但是上述的公式都是用理论值参与运算的


我们向上取整后，真实的失误率可以由如下公式得到：

公式3：

```math
p = (1 - e ^\frac{-nk}{m}  )^k
```

> 加工hash函数的技巧，比如我们找到两个hash函数f函数和g函数。例如我们找md5当做f函数，ssh加密函数当成g函数。第一个hash函数可以是f() + g(), 第二个hash函数可以是f() + 2g(), 同理第三个f() + 3g()。我们可以得到无数多个hash函数，而且这么些hash函数几乎独立。wiki上有解释


> md5算出的hash值是一个字符串，我们可以把它当成26进制的数


> 布隆过滤器不支持删除


### 2.2.3 布隆过滤器的使用场景


#### HDFS

最著名的使用场景是在hdfs分布式文件系统中，有很多个小文件放着各种各样的数据，如何定位一个string在哪个文件里面？


HDFS会把各个小文件维度，每个小文件建立一个布隆过滤器。先看该string在各个小文件里面哪些小文件的布隆过滤器中是描黑的状态。反之如果该String在某个布隆过滤器中是描白的，那么该小文件内肯定不存在该String。描黑的小文件中有可能存在该String,接下来把描黑的小文件各个遍历一遍，去找这个String


> 经典的布隆过滤器是不支持删除的，但是强制支持删除可以对经典布隆过滤器进行改造。比如可以把两个bit当成一个位置，相应的m编程2m; 描黑一个点可以编程01，再描黑10，删除该描黑点，变为01，再删除00。这样改造可以支持删除


> 布隆过滤器的唯一目的，是为了节约空间


100亿的数据量，预期失误率万分之一以下，30G以内的布隆过滤器可以搞定。且预期失误率为十万分之六



# 3 一致性Hash

一致性hash，被称为google发的三篇影响业内的论文之一；


一致性Hash是为了解决，服务器端分布式存储的方案。在老的存储方案中，比如web收到页面请求存一条记录，我们业务层会把他存在数据库服务器上，该服务器可能是单台，也可能是分布式存储，轮训，hash的方式都可以。

分布式存储有一个问题，假设我们根据记录的hash值模3，去把数据均分到三台机器上。但是如果我们有一种范围查询，比如查id大于10小于20的记录，那么我们需要到三台机器上分别寻找，再mearge，称为mapRedux


### 分布式存储的问题

1、所以对于范围查询比较多的，最好还是存在单台机器上，但是数据量比较大，单台服务器负载过重，可以考虑分布式存储，范围查询时再mearge

2、比如我们用三台服务器来负载我们的底层存储，假设随着业务的增加，我们需要增加三台，现在用6台机器来负载。那么之前三台机器的数据的迁移代价是很高的。我们需要把三台里面的数据每一条算hash值，再模6，去再次分布到我们的六台服务器上，数据迁移的代价是全量的，相应的我们需要减机器的时候也会是相同的处理，这是个很突出的问题


3、现在来说我们的服务都是非常的弹性，一致性hash就是解决上面很突出的问题的。一致性hash既可以保证迁移代价很低，也可以保证新迁移的机器数量负载均衡


### 一致性hash的实现思路

1、在之前存储中，我们算出记录的hash，模上机器数，找到这条记录的归属。现在我们把hash值的结果想象成一个环，比如md5加密的范围是2的64次方减1。我们把0到2的64次方减1的范围想象成一个环

2、现在假设我们需要在三台机器上分布式存储数据，我们可以以三台机器的ip不同，或者三台机器的mac地址不同来确定三台机器的归属。比如我们按三台机器的ip

3、每台机器的ip，按照相同的hash函数，每台机器算出HashCode的值，一定对应在环上的某个点上。由于hashCode的离散型，三个点不能保证把环均分掉

4、先假设能均分掉这个环，新来的需要存储的记录算出hash值，一定落在环上的某个位置，该位置顺时针找到的第一台机器的位置，该机器就是它需要归属存储到的机器。

5、如果某台机器m1要下线，环上属于该机器上的hash域，直接迁移到顺时针的下一台机器m2上即可。如果要添加一台机器mk，比如这台机器算出的hash值在环上第一台机器m1和第二台机器的m2之前，在环上m1到mk的数据直接迁移给mk，m2之前的区域为m1到m2，现在减去m1到mk的范围上的数据即可


6、上述5步骤的实现，假设我们把m1和m2和m3算出的hashCode分别是4，100亿，200亿。排序后，我们把机器和hash值的映射存下来。业务层分发数据存储的时候，去查这个表，决定分发到哪里，表变化及时通知到业务层。比如业务层来一条数据，算出hashCode为8，那么我们找大于等于8的第一个机器位置，那么我们需要放在8到100亿范围的归属m2上（可以二分查找加速）


### 解决一致性Hash环的均分问题（负载均衡）

出现均分问题，有两个情况，第一个是初始机器如何保证均分环，第二个加机器减机器怎么保证再次均分该环？


> 不采用机器的ip去抢环。我们可以准备一张表，分配给m1机器1000个随机字符串，分配给m2也1000个字符串，同样分配1000个随机字符串给m3。然后这3000个字符串算Hash值去均分环，机器和字符串的关系是1000对应1的关系，比如来了某一个数需要存储，我们算出hash值需要分配给某一个字符串抢到环区域。那么我们再根据这个字符串找到背后的实际物理机器，可以类比m1的1000个红点，m2的1000绿点，m3的1000个黄点


> 我们新增物理节点，对应着新增加1000个字符串的hash值去抢环，该向哪个字符串夺取环区域和之前Ip的hash加入环的夺取方式相同，同样的删减机器也是一样。这样我们就可以实现负载均衡

> 数据量不到400亿，不需要考虑hash碰撞的问题，这3000个字符串算出的hash值是不可能碰撞的。即使碰撞，某一个hash值即是m1的某个字符串落到的位置红点，又是m2的的某个字符串落到的位置绿点，那么数据冗余两份分别存到m1和m2，也不会有大的影响



实质还是利用hash函数的离散性，可以理解1000个点是某一种味道的香水分子数。三种味道的香水喷到房间里面，闻起来是均匀的；


### 利用hash函数的离散性不仅可以实现负载均衡，也可以实现负载管理

m1机器性能超强，m2和m3分别是m1机器性能的一半，我们可以给m1分配2000个字符串的hash值去抢环，给m2和m3各分配1000个即可实现


### 一致性Hash的应用案例

亚马逊的物流仓库，是每种商品都通过一致性Hash分布在仓库中，用户选择商品的时候，可以很及时的返回


在北京市，一个用户通过手机搜索十公里范围内的所有餐饮店。可以把北京地图按照经纬度切分，比如以天安门为原点，每个商家注册外卖平台的时候，除了要填写老板，店名之类的信息为，还存一条经纬度的记录


# 4 并行算法和岛问题

假设上下为1相连，可以认为是相同的一片岛。斜着连不算

那么下面的图形中存在两个岛

```
0 1 0 1 0 1 0
1 1 0 1 1 1 0
0 1 1 0 1 0 0
0 0 0 0 0 0 0

```

单内存，单cpu如何解决，多核心cpu怎么解决？

> 解题思路1，可以利用感染函数，到了某个位置，如果该位置为1，把自己连同周围相连的1全部感染成2；对于n行m列，时间复杂度为O(N*M)。每个节点遍历一次，递归调用由于改了状态，下面节点的递归往上看，由于为2不再递归，同理其他位置也一样。实质每个位置看了5次

> 解题思路2，并行。方法1虽然时间复杂度已经很好了。但是如果中国的每一个平方厘米算作一个岛屿，这种方法就不行。

1、可以采用并行算法加并查集的解法，比如一片区域,实际上只有一个岛，我们按照*号分割成两部分，分别让不同的CPU去跑感染函数


2、每个CPU跑完各自的区域后，记录岛的数量相加，和以*号分割的位置

3、通过并查集合并，如果处在*位置的左侧区域的点的右侧位置没被CPU2记录，那么不用管。如果CPU2记录了，那么需要合并，岛的总数减1

CPU1记录A和B，岛的数量为2。

CPU2记录C和D，岛的数量为2，一共4个岛


A和C并，岛数量变为3，集合变为(A B)；B和C并，岛数量为2，集合变为(A B C)；D和B可并，岛数量变为1，集合变为(A B C D)；全部连通


> 有了上述的思想，我们可以把岛切分成任意块，用不同的CPU去跑，然后再用并查集合并

**并行的思想，在大数据领域很重要**


```
1 1 1 1 1(A点) * (C点) 1 1 1 1
1 0 0 0 0      *      0 0 0 1
1 0 1 1 1(B点) * (C点) 1 1 1 1
1 0 1 0 0      *      0 0 0 0
1 0 1 1 1(B点) * (D点) 1 1 1 1
1 0 0 0 0      *      0 0 0 1
1 1 1 1 1(A点) * (D点) 1 1 1 1
```

```Go
package main

import "fmt"

func countIslands1(m [][]int) int {
	if len(m) == 0 || len(m[0]) == 0 {
		return 0
	}

	N := len(m)
	M := len(m[0])
	res := 0
	// 遍历
	for i := 0; i < N; i++ {
		for j := 0; j < M; j++ {
			// 如果该位置是1，调用感染函数，岛的数量增加1
			if m[i][j] == 1 {
				res++
				infect(m, i, j, N, M)
			}
		}
	}
	return res
}

// 感染函数
func infect(m [][]int, i, j, N, M int) {
	// 行，列越界，或者当前位置不是1，直接返回
	if i < 0 || i >= N || j < 0 || j >= M || m[i][j] != 1 {
		return
	}

	// 把当前位置1感染成2
	m[i][j] = 2
	// 检查四周的1，把四周的1也感染
	infect(m, i + 1, j, N, M)
	infect(m, i - 1, j, N, M)
	infect(m, i, j + 1, N, M)
	infect(m, i, j - 1, N, M)
}


// 岛问题，并行+并查集解法
func countIslands2(m [][]int) int {
	list := make([]string, 0)
	for row := 0; row < len(m); row++ {
		for col := 0; col < len(m[0]); col++ {
			if m[row][col] == 1 {
				position := fmt.Sprintf("%d_%d", row, col)
				list = append(list, position)
			}
		}
	}
	unionSet := InitUnionSet(list)
	for row := 0; row < len(m); row++ {
		for col := 0; col < len(m[0]); col++ {
			if m[row][col] == 1 {
				// row,col 5, 3 -> 5_3
				position := fmt.Sprintf("%d_%d", row, col)
				if row - 1 >= 0 && m[row - 1][col] == 1 {
					up := fmt.Sprintf("%d_%d", row - 1, col)
					unionSet.Union(up, position)
				}
				if col - 1 >= 0 && m[row][col - 1] == 1 {
					left := fmt.Sprintf("%d_%d", row, col - 1)
					unionSet.Union(left, position)
				}
			}
		}
	}
	return unionSet.getSetNum()
}

//3
//3
//1
//1
func main() {
	m1 := [][]int{
		{ 0, 0, 0, 0, 0, 0, 0, 0, 0 },
		{ 0, 1, 1, 1, 0, 1, 1, 1, 0 },
		{ 0, 1, 1, 1, 0, 0, 0, 1, 0 },
		{ 0, 1, 1, 0, 0, 0, 0, 0, 0 },
		{ 0, 0, 0, 0, 0, 1, 1, 0, 0 },
		{ 0, 0, 0, 0, 1, 1, 1, 0, 0 },
		{ 0, 0, 0, 0, 0, 0, 0, 0, 0 },
	}
	fmt.Println(countIslands1(m1))

	m1Other := [][]int{
		{ 0, 0, 0, 0, 0, 0, 0, 0, 0 },
		{ 0, 1, 1, 1, 0, 1, 1, 1, 0 },
		{ 0, 1, 1, 1, 0, 0, 0, 1, 0 },
		{ 0, 1, 1, 0, 0, 0, 0, 0, 0 },
		{ 0, 0, 0, 0, 0, 1, 1, 0, 0 },
		{ 0, 0, 0, 0, 1, 1, 1, 0, 0 },
		{ 0, 0, 0, 0, 0, 0, 0, 0, 0 },
	}
	fmt.Println(countIslands2(m1Other))

	m2 := [][]int{
		{ 0, 0, 0, 0, 0, 0, 0, 0, 0 },
		{ 0, 1, 1, 1, 1, 1, 1, 1, 0 },
		{ 0, 1, 1, 1, 0, 0, 0, 1, 0 },
		{ 0, 1, 1, 0, 0, 0, 1, 1, 0 },
		{ 0, 0, 0, 0, 0, 1, 1, 0, 0 },
		{ 0, 0, 0, 0, 1, 1, 1, 0, 0 },
		{ 0, 0, 0, 0, 0, 0, 0, 0, 0 },
	}
	fmt.Println(countIslands1(m2))

	m2Other := [][]int{
		{ 0, 0, 0, 0, 0, 0, 0, 0, 0 },
		{ 0, 1, 1, 1, 1, 1, 1, 1, 0 },
		{ 0, 1, 1, 1, 0, 0, 0, 1, 0 },
		{ 0, 1, 1, 0, 0, 0, 1, 1, 0 },
		{ 0, 0, 0, 0, 0, 1, 1, 0, 0 },
		{ 0, 0, 0, 0, 1, 1, 1, 0, 0 },
		{ 0, 0, 0, 0, 0, 0, 0, 0, 0 },
	}
	fmt.Println(countIslands2(m2Other))
}

// UNode 并查集结构中的节点类型
type UNode struct {
	V string
}

type UnionSet struct {
	// 记录样本到样本代表点的关系。值到代表该值的Node节点的关系映射
	Nodes map[string]*UNode
	// 记录某节点到根祖宗节点的关系。
	// 比如b指向a，c指向a，d指向a，a指向自身
	// map中保存的a->a b->a c->a d->a
	RootFatherMap map[*UNode]*UNode
	// 只有当前点，他是代表点，会在sizeMap中记录该代表点的连通个数
	SizeMap map[*UNode]int
}

// InitUnionSet 初始化一个并查集结构
func InitUnionSet(values []string) *UnionSet {
	us := &UnionSet{}
	nodes := make(map[string]*UNode, 0)
	fatherMap := make(map[*UNode]*UNode, 0)
	sizeMap := make(map[*UNode]int, 0)
	for _, v := range values {
		node := &UNode{V: v}
		nodes[v] = node
		fatherMap[node] = node
		sizeMap[node] = 1
	}

	us.Nodes = nodes
	us.RootFatherMap = fatherMap
	us.SizeMap = sizeMap
	return us
}

// FindFather 在并查集结构中找一个节点的父亲根节点
// 从点cur开始，一直往上找，找到不能再往上的代表点，返回
// 通过把路径上所有节点指向最上方的代表节点，目的是把findFather优化成O(1)的
func (set *UnionSet) FindFather(cur *UNode) *UNode {
	// 在找father的过程中，沿途所有节点加入当前容器，便于后面扁平化处理
	path := make([]*UNode, 0)
	// 当前节点的父亲不是指向自己，进行循环
	for cur != set.RootFatherMap[cur] {
		path = append(path, cur)
		// 向上移动
		cur = set.RootFatherMap[cur]
	}
	// 循环结束，cur此时是最上的代表节点
	// 把沿途所有节点拍平，都指向当前最上方的代表节点
	for len(path) != 0 {
		for i := len(path) - 1; i >= 0; i-- {
			set.RootFatherMap[path[i]] = cur
			path = path[:len(path) - 1] // 模拟栈的弹出
		}
	}
	return cur
}

// IsSameSet 判断两个元素是否在同一个并查集中
func (set *UnionSet) IsSameSet(a, b string) bool {
	// 先检查a和b有没有登记
	if _, ok := set.Nodes[a]; !ok {
		return false
	}
	if _, ok := set.Nodes[b]; !ok {
		return false
	}

	// 比较a的最上的代表点和b最上的代表点
	return set.FindFather(set.Nodes[a]) == set.FindFather(set.Nodes[b])
}

// Union 合并两个元素
func (set *UnionSet) Union(a, b string) {
	// 先检查a和b有没有都登记过
	if _, ok := set.Nodes[a]; !ok {
		return
	}
	if _, ok := set.Nodes[b]; !ok {
		return
	}

	// 找到a的最上面的代表点
	aHead := set.FindFather(set.Nodes[a])
	// 找到b的最上面的代表点
	bHead := set.FindFather(set.Nodes[b])
	// 只有两个最上代表点内存地址不相同，需要union
	if aHead != bHead {
		// 由于aHead和bHead都是最上面的代表点，那么在sizeMap里可以拿到大小
		aSetSize := set.SizeMap[aHead]
		bSetSize := set.SizeMap[bHead]
		var big *UNode
		var small *UNode
		// 哪个小，哪个挂在下面
		if aSetSize >= bSetSize {
			big = aHead
			small = bHead
		} else {
			big = bHead
			small = aHead
		}

		// 把小集合直接挂到大集合的最上面的代表节点下面
		set.RootFatherMap[small] = big
		// 大集合的代表节点的size要吸收掉小集合的size
		set.SizeMap[big] = aSetSize + bSetSize
		// 把被吸收掉的小set删除掉
		delete(set.SizeMap, small)
	}
}

func (set *UnionSet) getSetNum() int {
	return len(set.SizeMap)
}
```