黑乌鸦

Map解析

3/27/2024 Go解析

# Map的简介

Map数据结构有多种多样的名称：哈希表，散列，字典，Map，键值对，这些都是广义上的Map数据类型。Map类型的数据结构在大多数的编程语言中都有支持，日常使用最多，面试最经常问到的是GO语言中的Map和Redis中的字典。

这是GO语言官方文档的描述：

Maps are a convenient and powerful built-in data structure that associate values of one type (the key) with values of another type (the element or value).

Map基本上是GO里面最经常，甚至所有技术组件，编程语言中最经常使用到的一个数据结构，它有一个显著的特点：

  • 性能非常优秀。O(1)的复杂度。
  • 通过键值对，可以实现一个元素与另一个元素的映射。

在实际使用中，我们可以用Map作为复杂数据的中间结构，进而实现多个数据的组合。

# 原理

Map的底层实现原理就是哈希（hash），它的用处非常广泛，我们后续的学习和工作中，会多次遇到。设计一个好的Map结构，必须要解决两个问题：

  • 哈希函数
  • 解决冲突

# 哈希函数

Hash算法的定义：将任意长度的二进制值转换为固定长度的二进制值。举个具体点的例子:MD5。不管文件包多大，最后都能转成一个固定长度的字符串。我们最常用到的哈希算法：取模。

评估一个哈希函数的优劣，要看下它产出的结果是否足够的均匀分布。越是分布的不均匀，越容易出现哈希冲突，越是降低整体性能。

# 冲突解决

目前常用的哈希冲突方案有:开放寻址法和拉链法。我们简单介绍下：

开放寻址法

开放寻址法的底层是一个一维数组，当我们执行取模后，找到对应的位置，如果当前位置已经存在数据了，就顺序往后找到一个可以用的位置，把数据放进去。查询的时候也是同样的道理，找到位置后，进行一次键值比较，不一样的话就依次往后寻找，直到找到对应的值，或者发现一个空位置就返回。开放寻址法中对性能影响最大的是装载因子，它是数组中元素的数量与数组大小的比值

开放寻址法的好处是结构简单，缺点是如果数据塞得太多，会严重降低性能。当装载因子等于1时，就相当于直面数组结构，完全失去了Map的意义。

拉链法

拉链法是最常用的解决冲突的办法，我们后续遇到的大多数Map场景都是通过拉链法解决的。相比开放寻址，拉链法的底层结构要复杂一些，在一维数组的基础上，增加了一层链表，或者你可以简单将其理解为二维数组。数组的第一维装哈希桶的编号，二维装同在一个桶内的数据。查询的时候先找到键值所在的桶，然后去逐个比较键值，找到桶内对应的数据。 理论上，为了保证性能，每个桶内的数据都在个位数。随着数据量增加，我们只需要扩容桶的数量就行。拉链法的优势：

  • 实际存储地址，也就是桶内的存储空间，可以随用随取，降低开销。
  • 可以通过扩容，降低数据量激增带来的性能损耗。 这里充分体现了计算机工程学的终极思想：如果一层中间件搞不定，那就再加一层。

# Map使用

# 键值对

Go语言的Map对于键值对是有要求的。GO语言里有值类型，引用类型等，有些类型是不能进行“==”操作的。对于Map而言，他的值类型，可以为任意类型，但是键类型是不允许出现函数类型，Map类型，切片类型的。

//正常操作
aMap := Map[string]int{
	"one":    1,
	"two":    2,
	"three":  3,
}
M1 := Map[struct{}]string{} //可以用，但最好别这样写。
M2 := Map[interface{}]string{} //可以用，但千万别这样写

//错误操作

//这三类编译器会直接报错，无效的映射键类型: 必须为键类型完全定义比较运算符 == 和 !=
M3 := Map[Map[string]int]int{}
M4 := Map[func()]int{}
M5 := Map[[]int]int{}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

为什么会出现这种情况？hash算法是无法保证百分百不重复的，Go处理哈希冲突的时候，需要再拿着键的值来进行一轮比较，这时候就需要键值的类型必须要能够进行“==”操作。

GoMap查询一个数据是否存在的流程：

img

# 常用操作

声明

var M1 Map[string]int  //不建议这样写
M2 := new(Map[string]int) //千万别这样写。gofmt会提示的
//以下方式都可以声明一个Map
M3:=Map[string]int{}
M4 := make(Map[string]int) //这是比较推荐的方式。

//声明好后可以直接用,不会有任何问题。
a := M3["str"]
delete(M3,"str")
M3["str"] = 1
a = M3["str"]

//M1 要单独说一下
a := M1["str"] 
delete(M1,"str")
M1["str"] = 1 //这里会报panic
a = M1["str"]
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

按照推荐的方式声明一个Map,读写操作不会有任何问题。但是操作一个值为nil的Map时，读操作,删除操作都不会有问题，写操作会Panic。

使用

M1 := make(Map[string]int)

//M1["str"] = 1
value,ok := M1["str"]// value 默认为Map 定义的值，ok 固定位 bool 值
if ok {
	fmt.Printf("str:%d",value)
}
//可以合并：
if value,ok := M1["str"];ok {
	//注意此处 value ok 的作用域
	fmt.Printf("str:%d",value)
}
1
2
3
4
5
6
7
8
9
10
11
12

遍历

GoMap的遍历是随机的，无序的。

M1 := make(Map[int]int)

//正序塞入数据
M1[1] = 1
M1[2] = 2
M1[3] = 3
M1[4] = 4
M1[5] = 5

for k, v := range M1 {
	fmt.Println("key:%d,value:%d",k,v) //不是每一次都是排好序的
}
1
2
3
4
5
6
7
8
9
10
11
12

# Map的底层设计

  • 在Map的实现中，哈希的低位用来参与计算桶的编号，高8位会冗余一份存在桶内，用来加快比较。
  • 两个状态下会发生扩容：装载因子超过6.5；使用了大量的溢出桶。
  • 每个桶里只能存8个KV对，超过这个数量会分配在溢出桶内。

第一条在工作中经常会用这个方法来解决实际中的工程问题：比如记录用户信息和设备信息之间的对应关系。

# go map 概述

  • map 只是一个哈希表。数据被排列成一组 bucket。
  • 每个 bucket 最多包含 8 个 键/值 对。
  • 哈希值的低位字节位用于选择 bucket。
  • 每个 bucket 包含每个哈希的几个高位字节位(tophash)，以区分单个桶中的条目。
  • 如果超过 8 个 key 哈希到同一个桶，我们将额外的桶以链表的方式链接起来。（解决哈希冲突，链表法）
  • 当哈希表扩容时，我们会分配一个两倍大的新 bucket 数组。然后 bucket 从旧 bucket 数组增量复制到新 bucket 数组。
  • map 迭代器遍历 bucket 数组，并按遍历顺序返回键（遍历完普通桶之后，遍历溢出桶）。
  • 为了保持迭代语义，我们永远不会在它们的桶中移动键（bucket 内键的顺序在扩容的时候不变。如果改变了桶内键的相对顺序，键可能会返回 0 或 2 次）。
  • 在扩容哈希表时，迭代器仍在旧的桶中迭代，并且必须检查新桶，检查正在迭代的 bucket 是否已经被迁移到新桶。

# go map 的整体模型

上面讲了哈希表的基本设计思路，接下来就要开始讲 go 里面 map 的设计与实现了。大体上其实就是上面说的样子，但是有下面几个不一样的地方：

下文的 bucket 和 bmap 都是指的 "桶"。

  • go map 里面存储数据的地方是 bucket（桶），一个 bucket 可以存储 8 个元素，也就是说哈希冲突的时候还是会在同一个 bucket 中先存储。
  • 如果 bucket 也存放不下冲突的元素了，那么会新建另外一个桶（叫做溢出桶），旧的 bucket 记录这个新桶的指针，旧的 bucket 存放不下的元素，会存放到这个溢出桶中。
  • 如果溢出桶还是存放不下，那么再新建一个溢出桶，链接到上一个溢出桶中。

也就是说，在 go 的 map 实现中，哈希计算出来的值决定了 key 应该存放在哪一个 bucket 中。

go map 的整体结构如下图： img

  • buckets 记录了保存 map 数据的所有 bucket（这种下文统一称为普通桶），go 中使用 bmap 这个结构体来表示 bucket，溢出桶也是使用 bmap 结构体表示。

  • 如果 bucket（普通桶）哈希冲突太多导致存放不下，会新建一个 bucket，在原来的 bucket 上会有一个指针记录新建 bucket 的地址，这个新 bucket 下文统一称为溢出桶。

  • 在创建 map 的时候，如果我们指定的容量比较大（B >= 4 的时候），那么会预创建一个溢出桶。

也就是说，go 中解决哈希冲突的链表法，链表上的每一个元素是一个 bucket。go map 的实现里面，一个 bucket 可以存放 8 个键值对。 上面的 bmap 的数据结构如下图： img

  • bmap 就是 bucket（桶），不管是普通的桶还是溢出的桶，都是使用 bmap 结构体表示。

  • bmap 中存储数据的方式有点特别，它先存储了 8 个 tophash 值，一个 tophash 的大小为 1 个字节，每一个 tophash 记录的是 bmap 中每一个元素的哈希值的最高的 8 bit。

  • 接下来是 bmap 存储的 8 个元素的 key，在 8 个 key 之后是 8 个 bmap 存储的值。我们会发现 key 和 value 的存储是分开的，而不是 key/value、key/value 这种方式。go 中这种分开存储的方式有一个好处是可以减少内存对齐的开销，从而更省内存。

  • 最后是 overflow（溢出桶），如果 bmap 存满了，那就会新建一个溢出桶来保存新的数据，通过在旧的 bmap 上记录指针来记录溢出桶。

tophash 的作用是，在哈希冲突的时候，在 bucket 内进行查找的时候，是需要在 bucket 内从第一个元素遍历到最后一个元素来查找的。如果 key 太大，直接比较 key 的话效率会比较低下，通过记录哈希值的高 8 位，我们就可以在 buckeet 内查找的时候，先比较哈希值的 前 8 位，这样一来，map 的效率受到 key 大小的影响就会比较小。当然哈希值的高 8 位有可能相同，在这种情况下，我们再比较一下 key 本身就可以确定 bucket 的那个槽（slot/cell）是否是我们正在查找的那一个 key。

# go map 相关数据结构

我们大部分内容是跟下面两个结构体打交道：

  • hmap: map 的数据结构，包含了 bucket 的指针、bucket 的数量、键值对的数量等信息。
  • bmap: 桶，存储 key/value 的地方
//go map 数据结构
type hmap struct {
   count     int            // map 的元素数量。调用 len(map) 的时候返回此值
   flags     uint8          // map 标记：iterator/oldIterator/hashWriting/sameSizeGrow
   B         uint8          // 指示了当前哈希表持有的 buckets 的数量（2^B 是 bucket 的数量）
   noverflow uint16         // 溢出桶的数量
   hash0     uint32         // 哈希种子，计算 key 的哈希的时候会传入哈希函数
   buckets   unsafe.Pointer // 指向 buckets 的数组，大小为 2^B。 如果元素个数为 0 则为 nil。

   // 哈希表扩容的时候记录 buckets 字段。
   // 增量扩容时，oldbuckets 的长度是 buckets 的一半。
   oldbuckets unsafe.Pointer

   nevacuate uintptr   // 指示扩容进度，小于此地址的 buckets 完成迁移
   extra     *mapextra // 可选字段
}

// mapextra 包含并非在所有 map 上都存在的字段。
// 下面 mapextra 注释是原文的翻译（看完了 map 的全部源码也还不是很懂这个结构体的作用，除了 nextOverflow 字段）。
type mapextra struct {
   // 如果 key 和 elem 都不包含指针并且是内联的，那么我们将 bucket type 标记为不包含指针。这避免了扫描此类 map。
   // 但是，bmap.overflow 是一个指针。 为了让溢出桶保持活动状态，
   // 我们将指向所有溢出桶的指针存储在 hmap.extra.overflow 和 hmap.extra.oldoverflow 中。
   // overflow 和 oldoverflow 仅在 key 和 elem 不包含指针时使用。
   // overflow 包含 hmap.buckets 的溢出桶。
   // oldoverflow 包含 hmap.oldbuckets 的溢出桶。
   // 间接允许在 hiter 中存储指向切片的指针。
   overflow    *[]*bmap
   oldoverflow *[]*bmap

   // nextOverflow 持有指向空闲溢出桶的指针。
   nextOverflow *bmap
}

// go map 中的 bucket 结构体（实际保存 key/value 的地方）
type bmap struct {
   // tophash 通常包含此 bucket 中每个键的哈希值的最高的 8 位（1 字节）。
   // 如果 tophash[0] ＜ minTopHash，则 tophash[0] 为桶已迁移状态。
   // 这是一个长度为 8 的数组，因为一个 bucket 只能存储 8 个元素。
   // tophash 存储的是每一个元素的键的哈希的高 8 位。
   //（通过比较不同键的哈希的高 8 位可以提高 bucket 内的查找性能，因为键可能很大）
   tophash [bucketCnt]uint8
   
   // 然后是 8 个键，然后是 8 个值。（这里的 8 是代码写死的）
   // 注意：将所有键放在在一起，然后将所有值放在一起使代码比交替的 key/elem/key/elem/… 复杂一些，
   // 但它允许我们消除填充（减少内存对齐导致的内存浪费），例如 map[int64]int8，
   // 这种如果使用 key/elem 的方式存储则需要浪费几个字节用来对齐。
   keys [bucketCnt]keytype // 8 个键
   values [bucketCnt]valuetype // 8 个值
   overflow *bmap // 指向溢出桶的指针
}
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

对于 map 的数据结构，需要特别说明的是，bmap 的源码中实际只包含了 tophash 字段，而后面的三个字段 keys/values/overflow都是在编译期间动态添加的。这是因为 map 中可能存储不同类型的键值对，所以键值对占据的内存空间大小只能在编译时进行推导。这样一来，最终的结果是，我们在 map 的源码中，访问 key 和 value 的时候都需要通过 bmap 的首地址加上偏移量来进行访问的。

比如获取 bucket 中第 i 个 key 的方式：

k := add(unsafe.Pointer(b), dataOffset+i*uintptr(t.keysize))
1

这行代码中，add 是做指针加法运算的函数，具体来说就是第一个参数的地址加上第二个参数（偏移量），得到一个我们想要的指针。 dataOffset 代表了 bmap 的 keys 第一个元素的偏移量，i 代表了我们想要获取的 key 在 keys 中的索引：

这样一来，我们就可以通过 k 这个指针来访问 bucket 中的 key 了。同样的，要访问 value 的方式也是类似的，只要将 dataOffset + i * uintptr(t.keysize) 替换成 dataOffset + bucketCnt * uintptr(t.keysize) 即可。

这种方式虽然不太优雅，但是在性能上可以达到最优。

# bmap（桶）源码剖析

bmap 就是保存键值对的地方，但是它本身的方法并不多：

// bucket 是否已经完成迁移
// b 是 bucket 的指针
func evacuated(b *bmap) bool {
   h := b.tophash[0]
   return h > emptyOne && h < minTopHash
}

// 获取 b 的溢出桶
func (b *bmap) overflow(t *maptype) *bmap {
   // bmap 数据结构的最后一个指针就是指向溢出桶的指针
   return *(**bmap)(add(unsafe.Pointer(b), uintptr(t.bucketsize)-goarch.PtrSize))
}

// 设置 b 的溢出桶
// bmap 数据结构的最后一个指针指向溢出桶
func (b *bmap) setoverflow(t *maptype, ovf *bmap) {
   *(**bmap)(add(unsafe.Pointer(b), uintptr(t.bucketsize)-goarch.PtrSize)) = ovf
}

// 获取 b 中保存 keys 的指针（指向了桶内的第一个 key）
func (b *bmap) keys() unsafe.Pointer {
   return add(unsafe.Pointer(b), dataOffset)
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

在 ev 中用到了两个常量，在 bucket 的 tophash 里面，会通过下面几个标志来记录桶里面槽的状态：

// 这个 cell 是空的，并且在更高的索引或溢出处没有更多的非空 cell。
emptyRest = 0
// 这个 cell 是空的
emptyOne = 1
// key/elem 有效。 entry 已被迁移到较大的哈希表的前半部分（扩容了）。
evacuatedX = 2
// 同上，但迁移到大的哈希表的后半部分（扩容了）。
evacuatedY = 3
// cell 是空的，**bucket** 已经被迁移了
evacuatedEmpty = 4
// 正常填充单元格的最小 tophash。
minTopHash = 5
1
2
3
4
5
6
7
8
9
10
11
12

为了跟正常的 tophash 区分开来，如果计算出来的 tophash 小于 minTopHash，会将计算出来的 tophash 加上 minTopHash：

// tophash 计算 hash 的 tophash 值。
// 这是一个字节的大小的。（hash 最高的 8 位）
func tophash(hash uintptr) uint8 {
   // top 本质上就是 hash 的前面 8 个字节（goarch.PtrSize*8 - 8，左移位数：指针的字节大小 - 8 字节）
   top := uint8(hash >> (goarch.PtrSize*8 - 8))
   if top < minTopHash {
      top += minTopHash
   }
   return top
}
1
2
3
4
5
6
7
8
9
10

这样一来，通过 tophash 这一个字节就可以记录桶里面槽的状态了，非常节省空间。

# 拓展引申

# 为什么Map的循环访问会出现乱序的情况？

循环Map时，会从一个随机的位置开始。 对map的for循环遍历，汇编代码里面重点处理Go map迭代循环的是两个runtime方法，如下：

  • runtime.mapiterinit
  • runtime.mapiternext

转换后

var hiter map_iteration_struct
for mapiterinit(type, range, &hiter); hiter.key != nil; mapiternext(&hiter) {
    index_temp = *hiter.key
    value_temp = *hiter.val
    index = index_temp
    value = value_temp
    original body
}
1
2
3
4
5
6
7
8

实际上编译器对于 slice 和 map 的循环迭代有不同的实现方式，并不是 for 一扔就完事了，还做了一些附加动作进行处理。而上述代码就是 for range map 在编译器展开后的伪实现

runtime.mapiterinit源码如下

func mapiterinit(t *maptype, h *hmap, it *hiter) {
	...
	it.t = t
	it.h = h
	it.B = h.B
	it.buckets = h.buckets
	if t.bucket.kind&kindNoPointers != 0 {
		h.createOverflow()
		it.overflow = h.extra.overflow
		it.oldoverflow = h.extra.oldoverflow
	}

	r := uintptr(fastrand())
	if h.B > 31-bucketCntBits {
		r += uintptr(fastrand()) << 31
	}
	it.startbucket = r & bucketMask(h.B)
	it.offset = uint8(r >> h.B & (bucketCnt - 1))
	it.bucket = it.startbucket
    ...

	mapiternext(it)
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

通过对 mapiterinit 方法阅读，可得知其主要用途是在 map 进行遍历迭代时进行初始化动作。共有三个形参，用于读取当前哈希表的类型信息、当前哈希表的存储信息和当前遍历迭代的数据

源码中fastrand的部分，这个方法名，它是一个生成随机数的方法。再看看上下文：

...
// decide where to start
r := uintptr(fastrand())
if h.B > 31-bucketCntBits {
	r += uintptr(fastrand()) << 31
}
it.startbucket = r & bucketMask(h.B)
it.offset = uint8(r >> h.B & (bucketCnt - 1))

// iterator state
it.bucket = it.startbucket
1
2
3
4
5
6
7
8
9
10
11

在这段代码中，它生成了随机数。用于决定从哪里开始循环迭代。更具体的话就是根据随机数，选择一个桶位置作为起始点进行遍历迭代

因此每次重新 for range map，见到的结果都是不一样的。那是因为它的起始位置根本就不固定！

runtime.mapiternext 源码

func mapiternext(it *hiter) {
    ...
    for ; i < bucketCnt; i++ {
		...
		k := add(unsafe.Pointer(b), dataOffset+uintptr(offi)*uintptr(t.keysize))
		v := add(unsafe.Pointer(b), dataOffset+bucketCnt*uintptr(t.keysize)+uintptr(offi)*uintptr(t.valuesize))
		...
		if (b.tophash[offi] != evacuatedX && b.tophash[offi] != evacuatedY) ||
			!(t.reflexivekey || alg.equal(k, k)) {
			...
			it.key = k
			it.value = v
		} else {
			rk, rv := mapaccessK(t, h, k)
			if rk == nil {
				continue // key has been deleted
			}
			it.key = rk
			it.value = rv
		}
		it.bucket = bucket
		if it.bptr != b { 
			it.bptr = b
		}
		it.i = i + 1
		it.checkbucket = checkbucket
		return
	}
	b = b.overflow(t)
	i = 0
	goto next
}
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

在上面 runtime.mapiterinit 所述，已经选定了起始桶的位置。接下来就是通过 mapiternext 进行具体的循环遍历动作。该方法主要涉及如下：

  • 从已选定的桶中开始进行遍历，寻找桶中的下一个元素进行处理
  • 如果桶已经遍历完，则对溢出桶 overflow buckets 进行遍历处理

总结

通过这一番分析，原因也很简单明了。就是 for range map 在开始处理循环逻辑的时候，就做了随机播种...

# 如何实现Map的有序访问？

  • 使用切片，将Key先排好序，再遍历逐个去Map中取Value。
  • 包装一个新的数据结构。链表，或者其他结构。

# 如何安全的使用Map?

Map本身是不安全的，非原子操作的。多个GoRoutine并发操作Map轻则数据混乱，重则直接panic。如果要在并发中使用Map，要么自己加读写锁sync.RWMutex，要么使用sync.Map。

# 应该优先考虑哪些类型作为字典的键类型？

原则上讲：求哈希和判等操作的速度越快，对应的类型就越适合作为键类型。 具体而言：优先选用数值类型和指针类型，其次使用长度固定的字符串，最好不要用高级类型（数组，结构体，接口等）除了效率比较低，还容易引发其他问题。

# Go语言中的Map如何扩容？

扩容的时机：装载因子超过一定的阈值或者使用了太多的溢出桶时。

扩容的规则：

  • 等量扩容 使用溢出桶太多的时候会进行等量扩容。申请和原来等量的内容，将原来的数据重新整理后，写入到新的内存中。可以简单的认为是一次内存整理，目的是提高查询效率。

  • 增量扩容 分成两步：第一步进入扩容状态，先申请一块新的内存，翻倍增加桶的数量，此时buckets指向新分配的桶，oldbuckets指向原来的桶。 第二步，重新计算老的桶中的哈希值在新的桶内的位置（取模或者位操作），将旧数据用渐进式的方式拷贝到新的桶中。

渐进式迁移分两块，一方面会从第一个桶开始，顺序迁移每一个桶，如果下一个桶已经迁移，则跳过。另一方面，当我们操作某一个桶的元素时，会迁移两个桶，进而保证经过一些操作后一定能够完成迁移。

当我们访问一个正在迁移的Map时，如果存在oldbuckets，那么直接去中oldbuckets寻找数据。当我们遍历一个正在迁移的Map时，新的和旧的就会遍历，如果一个旧的的桶已经迁移走了，那么就直接跳过，反正不在旧的就在新的里。Map遍历本身就是无序的。

# 如果没有等量扩容会出现什么问题？

随着溢出桶缓慢增长，有内存溢出的风险。