Go-map底层原理

字数统计: 4k字   |   阅读时长: 18分
阅读量

以下源码分析基于go1.14.13版本

一、Map底层原理

Golang的Map数据结构底层原理是哈希桶，解决冲突Hash的冲突使用的是开散列，也就是将具有相同关键码的元素用单链表连接起来

哈希桶的原理：哈希桶使用顺序表来存放Key链表的头结点，每一个Key有对应的链表。

如有以下数字[1,6,8,11,13,14]，hash(key)=value%5，存储在hash表如下图：

而Golang定义的哈希桶又有不同，我们在源码分析中会继续进行讲解

二、举个小栗子

该例子就是对map进行初始化、插入、查询、遍历、删除等简单操作

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
func main() {
	// 创建
	var str,strLen map[int]string
	str = make(map[int]string)
	strLen = make(map[int]string,10)
	// 插入
	str[0] = "zero"
	str[1] = "one"
	str[2] = "twe"
	str[3] = "three"
	strLen[100]="100"
	fmt.Println(strLen)
	// 查询
	value, ok := str[3]
	fmt.Println("Query:",value, ok)
	// 遍历
	for value,ok :=range str{
		fmt.Println("QueryAll:",value,ok)
	}
	// 删除
	delete(str,0)
	fmt.Println(str)
}

看汇编代码，对应链接的源码接口（也即CALL命令后runtime.XXX）

1
2
3
4
5
6
7
8
9
10
11
12
# 注：只截取部分关键汇编代码
go tool compile -N -l -S maptest.go  
		# str = make(map[int]string)
        0x004b 00075 (maptest.go:14)    CALL    runtime.makemap_small(SB)
        #strLen = make(map[int]string,10)
        0x0076 00118 (maptest.go:15)    CALL    runtime.makemap(SB)
        # str[0] = "zero" 插入/更新
        0x01f3 00499 (maptest.go:21)    CALL    runtime.mapassign_fast64(SB)
        # value, ok := str[3]  查找
        0x02d4 00724 (maptest.go:24)    CALL    runtime.mapaccess2_fast64(SB)
        # delete(str,0) 删除
        0x0665 01637 (maptest.go:31)    CALL    runtime.mapdelete_fast64(SB)

三、源码分析

1、Map结构体
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
// A header for a Go map.
type hmap struct {
	count     int // map的大小,len()返回这个值
	flags     uint8 // map的标记，hashWriting当前状态是否是正在写的情况。
	B         uint8  // 最多有2^B个桶
	noverflow uint16 // approximate number of overflow buckets; see incrnoverflow for details
	hash0     uint32 // 计算key哈希值的随机值，保证一个key在不同map中存放的位置是随机的
	buckets    unsafe.Pointer // 当前哈希桶有 2^B个数组
	// 内存扩展的时候使用。指向旧哈希桶首地址
    // 一半大小的之前的bucket数组，只有在growing过程中是非 nil
	oldbuckets unsafe.Pointer 
	nevacuate  uintptr        
	// 当 key 和 value 都可以 inline(无指针) 的时候，就会用这个字段
	extra *mapextra 
}

// 一个桶的描述
type bmap struct {
    // bucketCnt=8  表示一个桶里面可以存放8个key-Value值。 tophash指的是key对应的hash值的高8位
	tophash [bucketCnt]uint8
}

一个map的哈希桶数据结构如下。可以看到与普通的hash桶不一样的是，桶链里面的溢出桶(overflow)可以装固定大小的K-V值。且桶的大小是2的B次方，B值是由申请的map大小值和后续的扩容决定的。

2、初始化

初始化分2种情况

  • make的时候不指定map的size。如：make(map[int]int)
    创建map时，当没有指定size大小或size为0时，不会创建哈希桶，会在插入元素时创建，避免只申请不使用导致的效率和内存浪费

  • make的时候指定map的size。如：make(map[int]int,100)

    这个时候会按照提供的元素个数，找一个可以放得下这么多元素的 B 值，也就是确定bucket的个数。bucket的大小是2的整数(B)次幂，按照前人总结哈希桶大小约= size/ 6.5

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
// 没有指定map的大小，初始化一个hmap结构体，返回一个hmap类型
func makemap_small() *hmap {
	h := new(hmap)
	h.hash0 = fastrand() // 生成一个哈希种子
	return h
}

// make(map[int]int,10) 指定size的大小。这里的hint即是10
func makemap(t *maptype, hint int, h *hmap) *hmap {
	mem, overflow := math.MulUintptr(uintptr(hint), t.bucket.size)
	if overflow || mem > maxAlloc {
		hint = 0
	}

	// 初始化hmap
	if h == nil {
		h = new(hmap)
	}
	h.hash0 = fastrand() // 生成一个哈希种子

    // 找到一个满足的B值，也就是确定桶bucket的个数
	B := uint8(0)
	for overLoadFactor(hint, B) {
		B++
	}
	h.B = B

	// 分配初始的 hash table
	if h.B != 0 {
		var nextOverflow *bmap
		h.buckets, nextOverflow = makeBucketArray(t, h.B, nil)
		if nextOverflow != nil {
			h.extra = new(mapextra)
			h.extra.nextOverflow = nextOverflow
		}
	}

	return h
}
3、查找

查找流程：我们暂且B值设为4，则共有2^4=16个bucket。

  • 计算key对应的hash值

  • 取hash值的后4(B)位，这个值指定目标值所在的bucket的位置(共16个)

  • 取hash值的前8位，这个值指定的是目标值在某个overflow的哪个格子(共8个)

  • 找到确定位置，返回Key值对应的Value值。

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
52
53
54
55
56
57
58
59
// 查询Key对应的value值
func mapaccess2(t *maptype, h *hmap, key unsafe.Pointer) (unsafe.Pointer, bool) {
	....
	// map为空，或者元素数=0，直接返回未找到
	if h == nil || h.count == 0 {
		if t.hashMightPanic() {
			t.hasher(key, 0) 
		}
		return unsafe.Pointer(&zeroVal[0]), false
	}
	// 当前Map正在写，不能进行操作。只用一个flag来表示状态。并发不安全
	if h.flags&hashWriting != 0 {
		throw("concurrent map read and map write")
	}
	hash := t.hasher(key, uintptr(h.hash0)) // 计算hash值
	m := bucketMask(h.B) // 若B=4,m=1111
	// hash&m 的意思就是取hash的后m位 决定是在哪个桶。B指的是内存的位置
	b := (*bmap)(unsafe.Pointer(uintptr(h.buckets) + (hash&m)*uintptr(t.bucketsize))) 
	if c := h.oldbuckets; c != nil {
		if !h.sameSizeGrow() {
			// There used to be half as many buckets; mask down one more power of two.
			m >>= 1
		}
		oldb := (*bmap)(unsafe.Pointer(uintptr(c) + (hash&m)*uintptr(t.bucketsize)))
		if !evacuated(oldb) {
			b = oldb
		}
	}
	// 取出key对应的hash值的前8位 前8位确定桶中格子的位置
	top := tophash(hash)
bucketloop:
	for ; b != nil; b = b.overflow(t) { // 遍历对应桶的桶链
		for i := uintptr(0); i < bucketCnt; i++ { //循环桶的每个格子
			if b.tophash[i] != top { // 对比hash值是否匹配
				if b.tophash[i] == emptyRest {
					break bucketloop
				}
				continue
			}
            // 可能找到了对应的key值
            // key对应的内存位置
			k := add(unsafe.Pointer(b), dataOffset+i*uintptr(t.keysize))
			if t.indirectkey() {
				k = *((*unsafe.Pointer)(k))
			}
			if t.key.equal(key, k) { // 最终要比较K值是否全部一致
				// 取出对应的value值
				e := add(unsafe.Pointer(b), dataOffset+bucketCnt*uintptr(t.keysize)+i*uintptr(t.elemsize))
				if t.indirectelem() {
					e = *((*unsafe.Pointer)(e))
				}
                // 找到，返回对应的value值，和true
				return e, true
			}
		}
	}
	// 没找到，返回对应的类型的0值，和false
	return unsafe.Pointer(&zeroVal[0]), false
}
4、插入（更新）

理解了查询的流程，插入的流程很类似了。

插入流程：

  • 计算key的hash值，并使用后B位计算属于哪个bucket

  • 若当前空为空桶，则创建一个bucket；bucket不够，扩展bucket

  • 遍历对应bucket的桶链，使用高8位hash看该key值是否存在，若存在，则进行更新操作

  • 若不存在，则找到空位进行插入

  • 若已有的bucket没有找到空位，则创建新的bucket进行存储

  • 期间若达到hash扩展的条件，则进行hash扩张，再进行插入

    注意插入函数只将KEY的值插入，返回存储value值对应的内存地址。汇编来进行将value值存储在对应的内存空间。

    如下图可以表示插入的流程：

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
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
// 插入（更新）
func mapassign(t *maptype, h *hmap, key unsafe.Pointer) unsafe.Pointer {
	if h == nil {
		panic(plainError("assignment to entry in nil map"))
	}
	....
	// 如果map正在写，则报错
	if h.flags&hashWriting != 0 {
		throw("concurrent map writes")
	}
	// 计算key的hash值
	hash := t.hasher(key, uintptr(h.hash0))

	// 设置当前map的状态为hashWriting
	h.flags ^= hashWriting
	// 若当前桶为空，则创建
	if h.buckets == nil {
		// 分配第一个桶
		h.buckets = newobject(t.bucket) // newarray(t.bucket, 1)
	}

again:
	// 根据后B位计算属于哪个桶
	bucket := hash & bucketMask(h.B)
	if h.growing() {
		growWork(t, h, bucket)
	}
	// 找到对应桶存储的内存位置
	b := (*bmap)(unsafe.Pointer(uintptr(h.buckets) + bucket*uintptr(t.bucketsize)))
	// 取出key对应的高8位hash值
	top := tophash(hash)
	// 空闲位置
	var inserti *uint8
	var insertk unsafe.Pointer
	var elem unsafe.Pointer
bucketloop:
	// 遍历桶链 注意不是哈希桶
	for {
		// 遍历一个桶内元素，一个桶最多装8个k-v
		for i := uintptr(0); i < bucketCnt; i++ {
			// 若该key对应的hash值不是存储的hash值
			if b.tophash[i] != top {
				// 找到一个空的格子
				if isEmpty(b.tophash[i]) && inserti == nil {
					// 如果这个槽位没有被占，说明可以往这里塞 key 和 value
					inserti = &b.tophash[i] // inserti指的是桶内第几个格子
					insertk = add(unsafe.Pointer(b), dataOffset+i*uintptr(t.keysize)) // key的地址
					elem = add(unsafe.Pointer(b), dataOffset+bucketCnt*uintptr(t.keysize)+i*uintptr(t.elemsize)) // value的地址
				}
				if b.tophash[i] == emptyRest {
					break bucketloop
				}
				continue
			}
			// 该key可能已经存在 取出已存在的key
			k := add(unsafe.Pointer(b), dataOffset+i*uintptr(t.keysize))
			if t.indirectkey() {
				k = *((*unsafe.Pointer)(k))
			}
			// 全部的key值对比，若不相等则说明这key不存在
			if !t.key.equal(key, k) {
				continue
			}
			// already have a mapping for key. Update it.
			// key确实存在
			if t.needkeyupdate() {
				typedmemmove(t.key, k, key)
			}
			// value值的地址，插入value值由编译器来执行
			elem = add(unsafe.Pointer(b), dataOffset+bucketCnt*uintptr(t.keysize)+i*uintptr(t.elemsize))
			// 结束
			goto done
		}
		// bucket 的 8 个槽没有满足条件的能插入或者能更新的，去overflow里继续找
		ovf := b.overflow(t)
		if ovf == nil {
			break
		}
		// 赋值为链表的下一个元素，继续循环
		b = ovf
	}

	// 没有找到 key，分配新的空间
    // 如果触及内存扩张的条件，进行hash表的扩张。
    // 1、bucket不够，导致overflow过大，查找效率低
    // 2、overflow过大
	if !h.growing() && (overLoadFactor(h.count+1, h.B) || tooManyOverflowBuckets(h.noverflow, h.B)) {
		hashGrow(t, h)
		// hashGrow 的时候会把当前的 bucket 放到 oldbucket 里
		// 但还没有开始分配新的 bucket，所以需要到 again 重试一次
		// 重试的时候在 growWork 里会把这个 key 的 bucket 优先分配好
		goto again 
	}
	// 前面在桶里找的时候，没有找到能塞这个 tophash 的位置
	// 说明当前所有 buckets 都是满的，分配一个新的 bucket
	if inserti == nil {
		newb := h.newoverflow(t, b)
		inserti = &newb.tophash[0]
		insertk = add(unsafe.Pointer(newb), dataOffset)
		elem = add(insertk, bucketCnt*uintptr(t.keysize))
	}

	// 插入key
	if t.indirectkey() {
		kmem := newobject(t.key)
		*(*unsafe.Pointer)(insertk) = kmem
		insertk = kmem
	}
	// 插入key
	if t.indirectelem() {
		vmem := newobject(t.elem)
		*(*unsafe.Pointer)(elem) = vmem
	}
	typedmemmove(t.key, insertk, key)
	*inserti = top
	// map的个数++
	h.count++

done:
	if h.flags&hashWriting == 0 {
		throw("concurrent map writes")
	}
	h.flags &^= hashWriting
	if t.indirectelem() {
		elem = *((*unsafe.Pointer)(elem))
	}
	// 注意这里返回的是k对应的value存储的内存的位置
	// 这里把key放在内存，什么时候放value呢，这里交给编译器来存储
	return elem
}

5、删除

删除流程：

  • 计算key对应的hash值
  • 取hash值的后4(B)位，这个值指定目标值所在的bucket的位置(共16个)
  • 取hash值的前8位，这个值指定的是目标值在某个overflow的哪个格子(共8个)
  • 找到确定位置。若key值是指针，则置空；否则
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
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
// 删除某个key对应的值
func mapdelete(t *maptype, h *hmap, key unsafe.Pointer) {
    ....
	if h == nil || h.count == 0 {
		if t.hashMightPanic() {
			t.hasher(key, 0) 
		}
		return
	}
	if h.flags&hashWriting != 0 {
		throw("concurrent map writes")
	}
	// 计算hash值
	hash := t.hasher(key, uintptr(h.hash0))

	h.flags ^= hashWriting
	// 计算在哪个桶
	bucket := hash & bucketMask(h.B)
	if h.growing() {
		growWork(t, h, bucket)
	}
	// 找到对应桶的内存位置
	b := (*bmap)(add(h.buckets, bucket*uintptr(t.bucketsize)))
	bOrig := b
	// key对应hash的高8位
	top := tophash(hash)
search:
	for ; b != nil; b = b.overflow(t) { // 桶链遍历
		for i := uintptr(0); i < bucketCnt; i++ { // 桶里面的8个格子遍历
			if b.tophash[i] != top {
				if b.tophash[i] == emptyRest {
					break search
				}
				continue
			}
			// 找到key的内存地址
			k := add(unsafe.Pointer(b), dataOffset+i*uintptr(t.keysize))
			k2 := k
			if t.indirectkey() {
				k2 = *((*unsafe.Pointer)(k2))
			}
			// 具体指比较
			if !t.key.equal(key, k2) {
				continue
			}
			// Only clear key if there are pointers in it.
			// 如果 key 中是指针，那么清空 key 的内容
			if t.indirectkey() {
				*(*unsafe.Pointer)(k) = nil
			} else if t.key.ptrdata != 0 { // 否则置为对应类型的0值
				memclrHasPointers(k, t.key.size)
			}
			// 计算value的内存地址
			e := add(unsafe.Pointer(b), dataOffset+bucketCnt*uintptr(t.keysize)+i*uintptr(t.elemsize))
			if t.indirectelem() {
				*(*unsafe.Pointer)(e) = nil
			} else if t.elem.ptrdata != 0 {
				memclrHasPointers(e, t.elem.size)
			} else {
				memclrNoHeapPointers(e, t.elem.size)
			}
			// 设某个对应的格子为空位置
			b.tophash[i] = emptyOne

			if i == bucketCnt-1 {
				if b.overflow(t) != nil && b.overflow(t).tophash[0] != emptyRest {
					goto notLast
				}
			} else {
				if b.tophash[i+1] != emptyRest {
					goto notLast
				}
			}
			for {
				b.tophash[i] = emptyRest
				if i == 0 {
					if b == bOrig {
						break // beginning of initial bucket, we're done.
					}
					// Find previous bucket, continue at its last entry.
					c := b
					for b = bOrig; b.overflow(t) != c; b = b.overflow(t) {
					}
					i = bucketCnt - 1
				} else {
					i--
				}
				if b.tophash[i] != emptyOne {
					break
				}
			}
		notLast:
			// 个数--
			h.count--
			break search
		}
	}

	if h.flags&hashWriting == 0 {
		throw("concurrent map writes")
	}
	h.flags &^= hashWriting
}
6、扩容

扩容发生在插入的mapassign中：

触发扩容的2种情况

1、几乎所有桶要满了，如果插入新元素，大概率挂在overflow上。也即平均每个bucket存储的键值对达到6.5个

解决办法：增量扩容。将B+1，也就是扩张bucket 1倍

2、overflow的桶太多了

  • bucket 总数 < 2 ^ 15 时

    overflow 的 bucket 总数 >= bucket 的总数，即认为overflow 的桶太多了

  • bucket 总数 >= 2 ^ 15 时

    overflow 的 bucket >= 2 ^ 15 时，即认为溢出桶太多了。

因为我们对 map 一边插入，一边删除，会导致其中很多桶出现空洞，这样使得 bucket 使用率不高，值存储得比较稀疏。在查找时效率会下降。

解决办法：等量扩容。通过移动 bucket 内容，使其倾向于紧密排列从而提高 bucket 利用率

1
2
3
4
5
6
7
8
9
10
11
func mapassign(t *maptype, h *hmap, key unsafe.Pointer) unsafe.Pointer {
    ....
	if !h.growing() && (overLoadFactor(h.count+1, h.B) || tooManyOverflowBuckets(h.noverflow, h.B)) {
		hashGrow(t, h)
		// hashGrow 的时候会把当前的 bucket 放到 oldbucket 里
		// 但还没有开始分配新的 bucket，所以需要到 again 重试一次
		// 重试的时候在 growWork 里会把这个 key 的 bucket 优先分配好
		goto again 
	}
    ....
}
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
// hash扩容
func hashGrow(t *maptype, h *hmap) {
	// 如果已经超过了 load factor 的阈值，那么需要对 map 进行扩容，即 B = B + 1，bucket 总数会变为原来的二倍
	// 如果还没到阈值，那么只需要保持相同数量的 bucket，横向拍平就行了
	bigger := uint8(1)
	if !overLoadFactor(h.count+1, h.B) {
		bigger = 0
		h.flags |= sameSizeGrow
	}
	oldbuckets := h.buckets
	newbuckets, nextOverflow := makeBucketArray(t, h.B+bigger, nil)

	flags := h.flags &^ (iterator | oldIterator)
	if h.flags&iterator != 0 {
		flags |= oldIterator
	}
	// 提交扩容结果
	h.B += bigger
	h.flags = flags
	h.oldbuckets = oldbuckets
	h.buckets = newbuckets
	h.nevacuate = 0
	h.noverflow = 0

	if h.extra != nil && h.extra.overflow != nil {
		// 把当前的 overflow 赋值给 oldoverflow
		if h.extra.oldoverflow != nil {
			throw("oldoverflow is not nil")
		}
		h.extra.oldoverflow = h.extra.overflow
		h.extra.overflow = nil
	}
	if nextOverflow != nil {
		if h.extra == nil {
			h.extra = new(mapextra)
		}
		h.extra.nextOverflow = nextOverflow
	}
}

下面详细说明2种扩容的方法：

1、增量扩容

  • oldbuckets指向当前的bucket

  • 开始扩容，B=B+1。

  • buckets指向新申请的bucket

  • 将老数据迁移到新指向的bucket

  • 删除oldbuckets指向的bucket

    场景如下：

    1） 触发增量扩张，只有一个bucket，且overflow装了7个元素(B=1)

​ 2）扩张bucket。B=2，oldbuckets指向当前的bucket。buckets指向新申请的bucket

3）数据迁移。删除oldbuckets指向当前的bucket

2、等量扩容

也就是将松散的键值对重新排列一次，以使bucket的使用率更高，进而保证更快的存取。将空的格子放在一起

三、思考

1、是协程安全的吗？

不是，可以看到map的状态是以hmap的flag字段来标识，在并发状态情况下，该字段不会被保护。会出现读写错误甚至是死锁的情况。

解决办法：

加锁，每次读写操作都加锁。但次锁的粒度是整个maps

使用sync.map。这个库的map已经包含读写锁。使用于大量读、少量写

2、申请的是什么类型的内存

堆内存。删除掉map中的元素不会释放内存，删除操作仅仅将对应的tophash[i]设置为empty，并非释放内存。若要释放内存只能等待指针无引用后被系统GC。

四、引用

赏

感谢您的认可

支付宝

扫一扫，分享到微信

微信分享二维码

tag:

    缺失模块。
    1、请确保node版本大于6.2
    2、在博客根目录（注意不是yilia根目录）执行以下命令：
    npm i hexo-generator-json-content --save

    3、在根目录_config.yml里添加配置： 

      jsonContent:
    meta: false
    pages: false
    posts:
      title: true
      date: true
      path: true
      text: false
      raw: false
      content: false
      slug: false
      updated: false
      comments: false
      link: false
      permalink: false
      excerpt: false
      categories: false
      tags: true