基本原理

go中的map底层用如下结构表示：

type hmap struct {// 元素数量count     int // 表示map的一些状态，例如是否在被迭代，是否在写flags     uint8// 桶个数的对数，即 2^B = cap(buckets)B         uint8  // 溢出桶的数量noverflow uint16// 计算key的hash值时使用的hash种子hash0     uint32 // hash seed// 指向buckets数组，大小为 2^Bbuckets    unsafe.Pointer // 扩容时指向老的buckets数组oldbuckets unsafe.Pointer // 指示扩容进度nevacuate  uintptr      extra *mapextra // optional fields
}

桶的结构

buckets数组的长度一定是2的B次幂，这样就能用位运算的方式高效对key的hash值取模

其指向的是一个bmap结构体数组。bmap定义如下：

type bmap struct {tophash [bucketCnt]uint8
}

编译时会将其完善成如下形式：

type bmap struct {tophash  [8]uint8keys     [8]keysizevalues   [8]valuesizeoverflow uintptr
}

• 为啥需要编译时完善？

  • 因为无法知道用户使用的map中，key和value的长度是多少，只能在编译期间，构造maptype时确定

• bmap就是具体的桶，最多存放8个键值对。对key的hash值按照桶数组的长度取模后，余数相同的键值对进入相同的桶

• 每个key的hash值高8位放到topbits的每个槽中，接下来8个位置存放key，8个位置存放value

• 当桶中装不下更多元素，且key又被hash到这个桶时，就会放到溢出桶，原桶的overflow字段指向溢出桶
  

这里引出两个问题：

• 为啥一个桶中放8个元素？因为8是一个合适的选择

  • 如果桶中能容纳的元素大于8个，则会导致大量key挤到一个桶中，导致在桶中查找变慢

  • 如果小于8个，则会消耗更多空间，且需要不断的去溢出桶中找，两个桶的空间不是一块分配的，因此大概率不在一个cpu cache中，效率也会降低

    • 例如java的hashMap是一个桶只存放一个键值对，沿着桶的next指针查找肯定不如就在一块连续的内容空间查找快

• 为啥8个key放一起，8个value放一起，而不是key/value的格式存放？

  • 源码里有说明这样的好处是节省内存对齐的空间
  • 例如map[int64]int8，如果按照 key/value/key/value/... 这样的模式存储，那在每一个 key/value 对之后都要额外 padding 7 个字节；而将所有的 key，value 分别绑定到一起，这种形式 key/key/.../value/value/...，则只需要在最后添加 padding

怎么定位key

对哈希表来说，除了桶的结构以外，最重要的就是如何根据key定位到桶，以及桶中的键值对

• key经过哈希计算后得到hash值，共 64 个 bit 位（这里以64位机器讨论）

• 计算它到底要落在哪个桶时，由最后 B 个 bit 位决定

  • B为hmap.B，桶数组容量的对数
  • 也就是说如果不同key的hash值如果最后B个bit位相同，会落到同一个桶中

• 定位到桶后，用hash的高8位去桶bmap的tophash数组中挨个匹配，如果匹配成功，且key和桶中对应位置的key相同，则定位到键值对

创建

在go代码中make(map[T]T)的操作，编译器会编译成对runtime.makemap的调用

该方法就是构造hmap结构，并根据初始容量预分配buckets

func makemap(t *maptype, hint int, h *hmap) *hmap {mem, overflow := math.MulUintptr(uintptr(hint), t.bucket.size)if overflow || mem > maxAlloc {hint = 0}if h == nil {h = new(hmap)}h.hash0 = fastrand()// 找到满足装count个元素最小的BB := uint8(0)for overLoadFactor(hint, B) {B++}h.B = Bif h.B != 0 {var nextOverflow *bmap// 分配buckets数组，和overflow桶，h.buckets, nextOverflow = makeBucketArray(t, h.B, nil)if nextOverflow != nil {h.extra = new(mapextra)h.extra.nextOverflow = nextOverflow}}return h
}

怎么确定B的大小？

func overLoadFactor(count int, B uint8) bool {return count > bucketCnt && uintptr(count) > loadFactorNum*(bucketShift(B)/loadFactorDen)
}

• bucketShift(B)：1 << B
• loadFactorNum/loadFactorDen：6.5

overLoadFactor方法判断count是否比 6.5* 2^B大，即2B个桶是否满足装count个数的需要，返回true表示不满足

创建map时干了以下几件事：

• 确定B的大小

• 分配桶数组空间，预分配溢出桶

  • 在make中设置了初始化容量才会分配桶数组空间，否则等到赋值时再分配
  • 当b>=4时，预分配 2^(b-4)个溢出桶，放到h.extra.nextOverflow中

查找

从map中查找元素，即go代码中的 v := map[key]，和v, ok := map[key]

其中v := map[key]形式的代码会被编译成调用 mapaccess系列的函数，我们先看 mapaccess1

1. h为nil，或者h中没有元素，返回对应类型的零值

if h == nil || h.count == 0 {return unsafe.Pointer(&zeroVal[0])
}

也就是说可以对值为nil的map执行查找操作，例如：

var m map[string]int
fmt.Println(m["0"]) // 0

2. 检测是否同时有其他协程正在写，如果是直接退出程序

if h.flags&hashWriting != 0 {throw("concurrent map read and map write")}

注意不是panic，而是throw，里面会调os.Exit退出程序，无法用recover捕获。因此在生产环境需要确保不能对原生map并发读写

3. 计算key的hash值

// 计算key的hash
hash := t.hasher(key, uintptr(h.hash0))

4. 定位到具体的桶b，做法为 hash & ((1 << h.b) - 1)，即hash与上桶的掩码

// m = (1 << h.B) - 1
m := bucketMask(h.B)
// b：定位到桶
b := (*bmap)(add(h.buckets, (hash&m)*uintptr(t.bucketsize)))

这顿操作的结果和 hash值对桶容量取余一样，但是效率更高

5. 如果正在扩容，且老桶还没搬迁，去老桶中找

// 如果正在扩容，且老桶还没搬迁，去老桶中找
if c := h.oldbuckets; c != nil {if !h.sameSizeGrow() {m >>= 1}oldb := (*bmap)(add(c, (hash&m)*uintptr(t.bucketsize)))if !evacuated(oldb) {b = oldb}
}

6. 计算tophash，即hash值的最高8位

top := tophash(hash)tophash函数如下，可以简单理解为右移56位
func tophash(hash uintptr) uint8 {top := uint8(hash >> (goarch.PtrSize*8 - 8))if top < minTopHash {top += minTopHash}return top
}

7. 接下来是一个双重for循环

   1. 第一层查找各个溢出桶
   2. 第二层在单个桶中查找每个槽位b.tophash[i]，看是否和top相同，如果不同，一定不是相同的key，跳过
   3. 如果相同，可能是相同的key，再比较key是否真的相等
   4. 如果key相同，返回value
   5. 当循环结束还没找到key时，返回对应类型的零值

bucketloop:for ; b != nil; b = b.overflow(t) {for i := uintptr(0); i < bucketCnt; i++ {if b.tophash[i] != top {if b.tophash[i] == emptyRest {break bucketloop}continue}// 定位到keyk := add(unsafe.Pointer(b), dataOffset+i*uintptr(t.keysize))if t.indirectkey() {k = *((*unsafe.Pointer)(k))}if t.key.equal(key, k) {// 定位到valuee := add(unsafe.Pointer(b), dataOffset+bucketCnt*uintptr(t.keysize)+i*uintptr(t.elemsize))if t.indirectelem() {e = *((*unsafe.Pointer)(e))}return e}}}return unsafe.Pointer(&zeroVal[0])
}

如何在桶中快速定位到key，value

定位到key：

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

dataOffset为第一个key相对于桶的偏移量，因此该语句能定位到第i个key的地址

定位到value：

e := add(unsafe.Pointer(b), dataOffset+bucketCnt*uintptr(t.keysize)+i*uintptr(t.elemsize))

在b的地址开始，跳过所有的key的位置，再跳过i个value的位置，就是第i个value的位置

第二种查找操作

v, ok := map[key]形式的go代码对应mapaccess2方法，方法签名如下：

func mapaccess2(t *maptype, h *hmap, key unsafe.Pointer) (unsafe.Pointer, bool) {

该方法整体和mapaccess1一样，只是多返回了一个bool，但查不到返回零值时，bool为false，其他情况bool为true

赋值

对map的赋值操作会被编译器编译为对mapassign系列函数的调用，我看看mapassign：

func mapassign(t *maptype, h *hmap, key unsafe.Pointer) unsafe.Pointer {// 往nil map赋值会panicif h == nil {panic(plainError("assignment to entry in nil map"))}// 不能同时写if h.flags&hashWriting != 0 {throw("concurrent map writes")}// 计算key的hash值hash := t.hasher(key, uintptr(h.hash0))// 将flag设置为正在写h.flags ^= hashWritingif h.buckets == nil {// 相当于 newarray(t.bucket, 1)h.buckets = newobject(t.bucket) }again:// 第几个桶bucket := hash & bucketMask(h.B)if h.growing() {// 帮助扩容一部分growWork(t, h, bucket)}// 定位到具体的桶b := (*bmap)(add(h.buckets, bucket*uintptr(t.bucketsize)))// 计算topHashtop := tophash(hash)var inserti *uint8var insertk unsafe.Pointervar elem unsafe.Pointer
bucketloop:for {for i := uintptr(0); i < bucketCnt; i++ {if b.tophash[i] != top {if isEmpty(b.tophash[i]) && inserti == nil {inserti = &b.tophash[i]insertk = add(unsafe.Pointer(b), dataOffset+i*uintptr(t.keysize))elem = add(unsafe.Pointer(b), dataOffset+bucketCnt*uintptr(t.keysize)+i*uintptr(t.elemsize))}if b.tophash[i] == emptyRest {break bucketloop}continue}k := add(unsafe.Pointer(b), dataOffset+i*uintptr(t.keysize))if t.indirectkey() {k = *((*unsafe.Pointer)(k))}if !t.key.equal(key, k) {continue}// already have a mapping for key. Update it.if t.needkeyupdate() {typedmemmove(t.key, k, key)}elem = add(unsafe.Pointer(b), dataOffset+bucketCnt*uintptr(t.keysize)+i*uintptr(t.elemsize))goto done}ovf := b.overflow(t)if ovf == nil {break}b = ovf}// 帮助扩容if !h.growing() && (overLoadFactor(h.count+1, h.B) || tooManyOverflowBuckets(h.noverflow, h.B)) {hashGrow(t, h)goto again // Growing the table invalidates everything, so try again}// 便利完了还没找到空的位置，说明需要新的桶if inserti == nil {// 给桶b分配溢出桶 newbnewb := h.newoverflow(t, b)// tophash位newb的第0个位置inserti = &newb.tophash[0]// key的位置insertk = add(unsafe.Pointer(newb), dataOffset)// value的位置elem = add(insertk, bucketCnt*uintptr(t.keysize))}// store new key/elem at insert positionif t.indirectkey() {kmem := newobject(t.key)*(*unsafe.Pointer)(insertk) = kmeminsertk = kmem}if t.indirectelem() {vmem := newobject(t.elem)*(*unsafe.Pointer)(elem) = vmem}// 将key放到insertK的位置typedmemmove(t.key, insertk, key)*inserti = toph.count++done:if h.flags&hashWriting == 0 {throw("concurrent map writes")}h.flags &^= hashWritingif t.indirectelem() {elem = *((*unsafe.Pointer)(elem))}return elem
}

整体赋值流程

1. 检查map是否为nil

   1. 往nil map赋值会panic

2. 不能并发写，否则会导致进程退出

3. 判断是否正常扩容，如果是帮助扩容

4. 定位到具体的桶，计算topHash

5. 和查找一样，赋值也有双重for循环，用到了3个变量：

   1. inserti：指向应该插入tophash的位置
   2. insertk：指向应该插入的key的位置
   3. elem：指向应该插入的value的位置

6. 在循环的过程中，inserti 和 insertk 分别指向第一个找到的空闲的tophash位置和key位置。如果之后在 map 没有找到 key ，也就是说原来 map 中没有此 key，这意味着插入新 key，则最终 key 的放置位置就是第一次发现的空闲位置

7. 如果跳出循环后发现inserti和insertk都是空，说明桶中的位置都满了，需要挂载溢出桶

   1. 溢出桶的空间可能实时分配，也可能从h.extra.nextOverflow中分配，因为创建map时会在h.extra.nextOverflow会存放一些预分配的溢出桶

如何赋值value

可以看到无论是更新还是插入，都值赋值了tophash和key，并没有对value赋值，那么value何时赋值呢？

假设go代码为：

func main() {var m = map[int]int{}m[1] = 2
}

汇编代码如下：

关于如何查看汇编代码，可以看：深入理解Go函数调用原理

MOVQ    "".m+192(SP), AX
MOVQ    AX, 8(SP)
MOVQ    $1, 16(SP)
CALL    runtime.mapassign_fast64(SB)
MOVQ    24(SP), AX
MOVQ    $2, (AX)

前三行分别在SP+0，SP+8，SP+16的地方设置了调用mapassign_fast64的参数，然后调用mapassign_fast64方法

根据函数调用规则，返回结果即elem的地址放在SP+24的位置

汇编第5行把SP+24赋值给AX

汇编第6行把立即数2赋值给AX指向的内存空间

总结：mapassign系列函数返回value的地址，汇编代码拿到往这个地址设置实际的值

key和value以什么形式放置在map中

mapassign中能看到如下的赋值代码：

if t.indirectkey() {kmem := newobject(t.key)*(*unsafe.Pointer)(insertk) = kmeminsertk = kmem
}
if t.indirectelem() {vmem := newobject(t.elem)*(*unsafe.Pointer)(elem) = vmem
}

如果是indirectkey，新开辟一片内存空间kmem，insertk指向kmem的地址，然后将key复制到kmem上，value同理

如果不是indirectkey，就将key直接放到map中

那怎么确定key是不是indirectkey呢？

// Note: flag values must match those used in the TMAP case
// in ../cmd/compile/internal/gc/reflect.go:dtypesym.
func (mt *maptype) indirectkey() bool { // store ptr to key instead of key itselfreturn mt.flags&1 != 0
}

mt.flags&1 == 1就是indirect，根据注释在gc /reflect.go找到如下代码：

if t.Key().Width > MAXKEYSIZE {ot = duint8(lsym, ot, uint8(Widthptr))flags |= 1 // indirect key
} // ...if t.Elem().Width > MAXELEMSIZE {ot = duint8(lsym, ot, uint8(Widthptr))flags |= 2 // indirect value
} // ...MAXKEYSIZE  = 128
MAXELEMSIZE = 128

答案揭晓，编译器在编译时就会根据key和value的size来决定是不是indirect，如果大于128就不将key或value放入map，而是新开辟一块内存存放

一般以string，或指针类型作为key都远远达不到128字节，就放在map里面

总结：当key，value为指针，或非指针但size小于128字节时放在map里，否则拷贝到一块新开辟的空间中，map中存放指向新空间的指针