[数据结构与算法]Go Map深度探究

Map相关问题

参考资料：https://www.cnblogs.com/qcrao-2018/archive/2019/05/22/10903807.html

Map初始化及使用

// 先声明，后使用
var m1 map[string]string
m1 = make(map[string]string)
// 最后给已声明的map赋值
m1["a"] = "aa"

// 直接创建
m2 := make(map[string]string)
m2["a"] = "aa"

// 初始化 + 赋值一体化
m3 := map[string]string{
	"a": "aa",
	"b": "bb",
}

// 查找键值是否存在
if v, ok := m1["a"]; ok {
	fmt.Println(v)
} else {
	fmt.Println("Key Not Found")
}

// 遍历map
for k, v := range m1 {
	fmt.Println(k, v)
}

// 删除key
delete(m1, "a")

hash函数如何确定？

hash函数分为两种，有加密型和非加密型。

加密型一般用来加密数据，典型代表就是md5、sha1、sha256、aes256.

非加密型一般用于查找，在map的应用场景中，就是这类。 选择hash函数主要考察的是两点：性能、碰撞概率。

map哈希函数的选择：

• 在程序启动时，检测cpu是否支持aes，如果支持，则使用aes hash，否则使用memhash。在alginit()函数中确定。

根据 key 的类型，_type 结构体的 alg 字段会被设置对应类型的 hash 和 equal 函数。

另外还会有一个hash0（哈希种子）来增加哈希函数的随机性。

map key的可比性

什么是Map？

Go中的map被称为哈希表。原理是将多个键/值对分散存储在buckets（桶）中。

哈希表通常会有一堆桶（buckets）来存储键值对，一个键值对来了，自然要选择一个桶，从到来到最后选择一个桶存储，要经历下面的过程：

1. Hash(k1) = hash. 可以通过hash函数把键处理一下，得到一个hash值。
2. 利用这个hash值从m个桶中选择一个，桶编号的区间就是[0,m-1]。

有两种方法比较常用——

取模法：用hash值与桶的个数取模。 hash % m，得到一个桶的编号

与运算法：hash & (m - 1)。 这种方法的m必须是2的整数次幂，比如2⁰，2¹等。

如果是整数次幂的情况，m-1就是低于这一位的所有位均为1.

这样hash值为xxxxxxxx的情况下，为0的都会被重置为0，为1的位与运算是它本身。 这样就可以辐射到所有的桶中。

如果不是整数次幂的情况下，m-1和hash与运算就无法辐射到下面的桶。

比如下面m=5，桶编号应该是[0,4]，然后下面m-1为00000100，任何数与它与运算只能得到0和4两个答案，所以[1,3]注定为空桶。

3. 桶选择好了之后，如果后来又有新的键值对选择了这个桶，就是发生了哈希冲突。解决哈希冲突的方法常用的有两种：

开放地址法：如果2号桶有使用的话，就往后再移动，索引加1。等下次找k2的时候，虽然还是会定位到2号桶，但是经过比较key不相等，会一直向后比较，直到key相等，或者遇到空桶，证明这个key不存在。

拉链法： 编号为2的桶被占用了，没关系，在它后面链一个新桶存储这个键值对。查找k2时，还是会先找到2号桶，经过比较发现key不相等，就会顺着链表向后查找，直到key相等，或者到了链表的尾部。

Go语言中是通过与运算法来确定桶的位置，通过拉链法来解决hash冲突。

在源码中，表示map的结构体是hmap，是hashmap的缩写：

// A header for a Go map.
type hmap struct {
	count     int  // 元素个数，调用 len(map) 时，直接返回此值
	flags     uint8  // 当前map的标志位，低四位来表示不同的状态。
	B         uint8  // buckets 的对数 
	noverflow uint16   // overflow 的 bucket 近似数
	hash0     uint32  // 计算 key 的哈希的时候会传入哈希函数
	buckets    unsafe.Pointer   // 指向 buckets 数组，大小为 2^B，如果元素个数为0，就为 nil
	oldbuckets unsafe.Pointer  // 扩容的时候，buckets 长度会是 oldbuckets 的两倍
	nevacuate  uintptr  	// 指示扩容进度，小于此地址的 buckets 迁移完成
	extra *mapextra 
}

flags的低四位分别表示如下：

在真正的使用中，只需要对某一位进行置1，置0操作即可。

iterator     = 1 // 可能有一个迭代器正在使用新桶
oldIterator  = 2 // 可能有一个迭代器正在使用旧桶
hashWriting  = 4 // 有一个goroutine正在写操作这个map
sameSizeGrow = 8 // 该map正在进行等量扩容

buckets数组中存放的是bmap结构体，bmap结构体就是我们所说的桶。

它大概的结构如下，首先buckets中存储的是bmap类型的数组地址（*bmap）。

bmap结构如右下角那样所示。

bmap的结构是：

type bmap struct {
    topbits  [8]uint8      // 通过hash值的前8位来确定存在哪个tophash中
    keys     [8]keytype    // 8个key
    values   [8]valuetype  // 8个value
    pad      uintptr
    overflow uintptr  // 溢出桶。 和bmap结构相同
}

通过hash值的高8位来确定是哪个tophash。

如果哈希表要分配的桶的数目大于2⁴，就认为使用溢出桶的几率较大，就会预分配2^(B-4)个溢出桶作备用。

hmap下有一个extra字段，用来记录溢出桶相关的信息。

overflow是一个slice，记录目前已经被使用的溢出桶的地址。

假如说编号为2的桶存满了，就会在后面链一个溢出桶，nextoverflow就会指向下一个空闲桶。

hmap中的noverflow记录使用的溢出桶的数量。

oldoverflow用于在扩容阶段存储旧桶用到的溢出桶的地址。

要查找一个key的流程图：

查询

源码：mapaccess1。

简单的说，假如要查询k1，首先通过key的hash值与运算得到桶的位置，如果是在扩容阶段，且这个bucket还没有迁移到新桶中，在oldbuckets中找，否则就在buckets中找，然后在桶中遍历tophash，与key的hash高八位进行比对，最终如果没有找到这个key值，返回类型零值，如果找到了，就返回这个key的value。

具体流程如下：

// 1. 如果map什么都没有，返回类型零值
if h == nil || h.count == 0 {
    if t.hashMightPanic() {
        t.hasher(key, 0) // see issue 23734
    }
    return unsafe.Pointer(&zeroVal[0])
}
// 2. 判断当前是否在写入，如果是在写入，抛出异常
if h.flags&hashWriting != 0 {
    throw("concurrent map read and map write")
}
// 3. 计算出当前key的hash值
hash := t.hasher(key, uintptr(h.hash0))
// 4. 通过hash值计算出桶的位置
b := (*bmap)(add(h.buckets, (hash&m)*uintptr(t.bucketsize)))
// 5. 判断是否在扩容阶段，如果不在扩容阶段，oldbuckets为nil，不进入下面的if语句。
if c := h.oldbuckets; c != nil {
    // 进来了，说明是处于扩容阶段，首先需要判断是否在等量扩容中
    if !h.sameSizeGrow() {
        // 如果不是等量扩容，就是翻倍扩容，使m右移一位，因为新bucket的数量是老bucket的两倍。需要使m除以2.
        m >>= 1
    }
    // 找到oldbucket的位置
    oldb := (*bmap)(add(c, (hash&m)*uintptr(t.bucketsize)))
    // evacuated用来判断这个bucket是否已经搬迁完毕，里面会有一些标志位。
    // 如果oldbucket没有搬迁到新的bucket中，那就在oldbucket中寻找
    if !evacuated(oldb) {
        b = oldb
    }
}
// 6. 计算出高 8 位的 hash
// 相当于右移 56 位，只取高8位
top := uint8(hash >> (sys.PtrSize*8 - 8))
// 7. 两层for循环遍历本bucket的所有tophash。
// 如果tophash不等于当前key的hash高八位，判断是否到了emptyRest，如果遇到了emptyRest，直接跳出循环。 否则继续探测下一个tophash。
if b.tophash[i] != top {
    if b.tophash[i] == emptyRest {
        break bucketloop
    }
    continue
}
// 当找到了tophash与key的hash高八位相同时，找到对应的key，并比对这个key是否和我们要找的key相同
// 定位到key的位置
k := add(unsafe.Pointer(b), dataOffset+i*uintptr(t.keysize))
if t.indirectkey() {
    // 解引用（得到key的值）
    k = *((*unsafe.Pointer)(k))
}
// 比对这个key和我们要找的key是否相同。 不相同就直接跳过
if t.key.equal(key, k) {
    // 定位到value的位置
    e := add(unsafe.Pointer(b), dataOffset+bucketCnt*uintptr(t.keysize)+i*uintptr(t.elemsize))
    // value解引用(得到value的值)
    if t.indirectelem() {
        e = *((*unsafe.Pointer)(e))
    }
    // 返回value
    return e
}
// 8. 如果没找到这个key，返回类型零值
return unsafe.Pointer(&zeroVal[0])

写入、修改

源码：mapassign

简单的说：假如我们现在来了一个key，还是先对key计算hash值，如果在扩容阶段，先将旧桶的这个bucket数据迁移到新桶（写时赋值，copy on write，只要在写入和修改的时候会进行迁移），然后通过两层循环遍历所有的tophash，和本key的hash高八位进行比对。

如果发现了key的高八位和tophash中的值相同，进行更新操作。 如果遍历了本bucket中的所有tophash，也没找到匹配的key的话，会在本bucket中遇到的第一个空的位置，存储进去，维护count。

具体流程如下：

// 1. 判断当前map是否为nil，如果为nil，报空指针
if h == nil {
    panic(plainError("assignment to entry in nil map"))
}
// 2. 判断当前map是不是在写状态中，如果别的goroutine在修改这个map，会报错。
if h.flags&hashWriting != 0 {
    throw("concurrent map writes")
}
// 3. 对当前的key计算hash值
hash := t.hasher(key, uintptr(h.hash0))
// 4. 修改当前map的写标志
h.flags ^= hashWriting
// 5. 如果在扩容阶段，将旧桶中这个bucket的数据迁移到新桶。 确保新桶中的bucket对应老bucket已经完成了迁移工作。
if h.growing() {
    growWork(t, h, bucket)
}
// 6. 计算出在新桶中的索引位置，以及hash的高八位
b := (*bmap)(unsafe.Pointer(uintptr(h.buckets) + bucket*uintptr(t.bucketsize)))
top := tophash(hash)
// 7.  双层循环遍历当前bucket的所有tophash。 一层遍历所有bmap，一层遍历bmap内的tophash。
// 会依次遍历当前bmap的所有tophash
// -----------如果当前的tophash不等于我们key的hash值高八位。----------------
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))
    }
    // 如果当前的位置为emptyRest，直接return，说明已经找了所有的tophash，没有找到与本key相同的值
    if b.tophash[i] == emptyRest {
        break bucketloop
    }
    // 继续下一个tophash
    continue
}
// -----------如果当前的tophash等于我们key的hash值高八位。----------------
// 首先找到这个tophash所对应的key
k := add(unsafe.Pointer(b), dataOffset+i*uintptr(t.keysize))
if t.indirectkey() {
    k = *((*unsafe.Pointer)(k))
}
// 如果发现这个key和我们要存的key不同，继续比对下一个tophash。 也就是两个key的哈希高八位是相同的，但是key却不一样，就将本tophash跳过。
k := add(unsafe.Pointer(b), dataOffset+i*uintptr(t.keysize))
if t.indirectkey() {
    k = *((*unsafe.Pointer)(k))
}
// 如果发现可以相同，发现map中已经有这个key了，进行更新操作。将其进行修改，然后维护map的flag标识。
if t.needkeyupdate() {
    typedmemmove(t.key, k, key)
}
elem = add(unsafe.Pointer(b), dataOffset+bucketCnt*uintptr(t.keysize)+i*uintptr(t.elemsize))
// 8. 进入溢出桶，不断的循环上面的操作，直到所有的都遍历完成，也没有跳出循环时，说明我们是要插入数据，而不是更新。
// 9. 判断当前是否处在扩容时机中，如果满足，进行扩容，并且从1开始重新进行操作（扩容会使一切无效，需要重新来一遍前面1-9的步骤）。
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
}
// 10. 判断inserti是否为nil，如果为nil意味着当前buckets是满的，需要添加一个新的
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))
}
// 11. 保存新的key/value，tophash
if t.indirectkey() {
    kmem := newobject(t.key)
    *(*unsafe.Pointer)(insertk) = kmem
    insertk = kmem
}
if t.indirectelem() {
    vmem := newobject(t.elem)
    *(*unsafe.Pointer)(elem) = vmem
}
typedmemmove(t.key, insertk, key)
*inserti = top
// 12. map count++
h.count ++
// 13. 维护map flags
h.flags &^= hashWriting

删除

源码在mapdelete函数。

简单的说，就是先对key进行hash，通过与运算计算出桶的编号，然后检测是否正在扩容过程，如果在扩容阶段，进行一次搬迁操作，否则的话直接进行tophash的遍历，找到对应的key，对key和value进行清零操作，维护count，然后将当前tophash的位置设置为emptyOne，接下来还要经过一些操作维护emptyRest。

具体的流程：

1. 先检查h.flags标志，如果发现写标志位为1，直接panic，表明有其它协程同时在进行写操作。
2. 检测是否正在扩容过程，直接触发一次搬迁操作，将本bucket搬迁到新buckets中。

bucket := hash & bucketMask(h.B)
if h.growing() {
    growWork(t, h, bucket)
}

3. 通过hash的高八位来找tophash

4. 一共有两层循环，一层循环bmap，一层循环tophash。 它会先比对当前tophash的值是否等于删除的key的hash高八位，如果不等于的话，判断一下当前位是不是emptyRest，如果是就代表未来已经没有key了，可以直接跳出循环，否则比对下一个tophash。

// 如果当前tophash的值不等于传进来的top值
if b.tophash[i] != top {
	// 如果遇到了emptyRest，说明已经找到了最后一个空值了，后面都是空值，直接跳出当前语句块即可。
	if b.tophash[i] == emptyRest {
		break search
	}
	// 继续比对下一个tophash
	continue
}

5. 找到对应位置，对key或者value进行“清零操作”，并将当前tophash设置为emptyOne。

// 对 key 清零
if t.indirectkey {
	*(*unsafe.Pointer)(k) = nil
} else {
	typedmemclr(t.key, k)
}

// 对 value 清零
if t.indirectvalue {
	*(*unsafe.Pointer)(v) = nil
} else {
	typedmemclr(t.elem, v)
}
// 将当前位设置为emptyOne
b.tophash[i] = emptyOne

6. 如果发现后面此元素后面没有元素，则需要将此tophash设置为emptyRest，并且还会循环向上检查前一个元素是否为空，如果为空，则将此元素设置为emptyRest。

// 下面的if else是判断当前元素的下一个是否是emptyRest
if i == bucketCnt-1 {
    // 此时i是我们找到的要删除的那个tophash的位置，如果i是在本bmap的最后一个位置
   // 我检测到我下面是否还有溢出桶，如果还有的话，看一下下一个元素是否是emptyRest
   // 如果是emptyRest，我就没必要再往下寻找了，直接跳到notLast，维护count的值，代表不是在最后。
    if b.overflow(t) != nil && b.overflow(t).tophash[0] != emptyRest {
        goto notLast
    }
} else {
    // 否则的情况下，就说明当前的i不是在当前bmap的最后一个位置
    // 直接判断当前bmap的下一个元素是不是emptyRest就可以了。
    // 如果不是，说明不是在最后，直接去notLast，维护count
    if b.tophash[i+1] != emptyRest {
        goto notLast
    }
}
// 这个时候，就是我们下一个元素是emptyRest的情况
for {
    // 当前位置现在是emptyOne，先把当前位置设置为emptyRest
    b.tophash[i] = emptyRest
    // 如果当前tophash索引为0
    if i == 0 {
        // 如果当前的bmap是bucket的第一个bmap
        if b == bOrig {
            // 直接break，因为已经遍历到第0号bmap的第0号tophash了，前面已经没有元素可以遍历了。
            break 
        }
        //  找到前一个bmap，然后把i变成这个bmap的最后一个tophash索引号
        // 首先使c存一下当前bmap
        c := b
        // b初始化为第一个bmap，然后不断的向溢出桶迭代
        // 当判断下一个bmap是c时跳出，意味着当前的bmap是c的上一个
        for b = bOrig; b.overflow(t) != c; b = b.overflow(t) {
        }
        // bucketCnt是tophash的长度。因为索引是从0开始的，所以i=bucketCnt - 1
        // 当前i就等于当前bmap的最后一个tophash的索引号了
        i = bucketCnt - 1
    } else {
        // 还在当前bmap tophash 中向前遍历。 维护索引号
        i--
    }
    // 如果当前下标的tophash是有值的，跳出当前循环。
    // 意味着如果当前tophash为emptyOne的话，也会变成emptyRest，然后继续向前滚动。
    if b.tophash[i] != emptyOne {
        break
    }
}
notLast:
	// 维护count
	h.count--
	// 跳出整个search语句块，代表删除操作完成。
	break search

扩容时机和扩容机制

渐进式扩容可以避免一次性扩容带来的暂时性能抖动

源码如下：

// 触发扩容的源码
// src/runtime/hashmap.go/mapassign

// 触发扩容时机
if !h.growing() && (overLoadFactor(int64(h.count), h.B) || tooManyOverflowBuckets(h.noverflow, h.B)) {
		hashGrow(t, h)
}

// 装载因子超过 6.5 map的元素个数除以桶的数量
func overLoadFactor(count int64, B uint8) bool {
	return count >= bucketCnt && float32(count) >= loadFactor*float32((uint64(1)<<B))
}

// overflow 的 bucket 数量超过 2^15。
func tooManyOverflowBuckets(noverflow uint16, B uint8) bool {
	if B < 16 {
		return noverflow >= uint16(1)<<B
	}
	return noverflow >= 1<<15
}

// growing reports whether h is growing. The growth may be to the same size or bigger.
func (h *hmap) growing() bool {
   return h.oldbuckets != nil
}

使用哈希表的目的就是希望快速的找到key，然而随着map中添加的key越来越多，key发生碰撞的概率也越来越大。最理想的情况下是一个bucket只装一个key，这样能达到O(1)的效率，但空间消耗太大，用空间换时间的代价太高。

Go语言采用一个bucket里装载8个key，定位到某个bucket之后，还需要再定位到具体的key，就是用时间换空间。不过，如果一直装在一个bucket之后，就退化成了链表。各种操作降为O(n)。

当需要扩容时，就要分配更多的桶，它们就是新桶，需要把旧桶里存储的键值对都迁移到新桶中，如果哈希表存储的键值对比较多，那么一次性迁移到新桶花费的时间就很显著。 所以通常哈希表扩容时，先分配足够多的新桶，然后用一个字段（oldbuckets）记录旧桶的开始位置，然后再用一个字段（nevacuate）记录要迁移的旧桶编号。

当哈希表每次读写操作时，如果检测到当前处于扩容阶段，就完成一部分键值对迁移任务，直到所有旧桶迁移完成，旧桶不再使用，才算真正完成一次哈希表的扩容。

这种把键值对迁移的时间分摊到多次哈希表操作中的方式，被称为“渐进式扩容”。

当遇到下面两种情况的时候会产生扩容：

1. 负载因子超过6.5，就会 产生翻倍扩容，新桶的数目是旧桶的两倍。 Load Factor = count / m。 键的数量/桶的数量。
2. 溢出桶太多会触发等量扩容。 当桶的数目小于等于2¹⁵时，溢出桶的数目大于等于常规桶的数目就代表多了。当桶的数目大于2¹⁵时，当溢出桶的数目大于等于2^15时，就相当于多了。

翻倍扩容

假如说，我们在一个桶中存了8个key，装载因子算出来就是8/1=8。 因此当装载因子超过6.5时，表明很多的bucket都快要装满了，这时查找和插入的效率都变低了，进行翻倍扩容，是有必要的。

当出现翻倍扩容时buckets指向新桶，oldbuckets指向旧桶，nevacuate=0，表示接下来要迁移编号为0的旧桶。

每个旧桶的键值对都会分流到两个新桶中。

以下面为例子，oldBuckets有四个桶，newBuckets有八个桶，假如说选择了旧桶的0号下标。

也就是hash & (m - 1)，m=4，则hash & (4 - 1) = 0。

也就是下图中的二进制表示情况，二进制低两位一定为0。

那么选择新桶的方式就只有两种，取决于哈希值的第三位是0还是1。

如果第三位为0，就迁到0号新桶。 如果第三位为1，就迁到4号新桶。

旧桶的每一个，就是这样分配到两个新桶中。

等量扩容

在装载因子比较小的情况下，也会导致查找和插入的效率降低，而第1点识别不出来这种情况。

也就是key的数量很小，但是bucket数量特别多。

造成这种情况的原因：不停地插入、删除元素，先插入很多元素，导致创建了很多的bucket，但是装载因子又没有达到临界值，未触发扩容缓解这种情况，又删除元素降低元素总数量，再插入很多元素，导致创建了很多的overflow bucket。

很多的溢出桶会导致key很分散，这就像是一座空城，房子很多，但是住户很少，找起人来很困难。

所以才会有这条规定，来进行等量扩容，只是把分散的住户拢一拢，聚集一些，但是房子还是那么多。

左边来看，key很分散，很多键值对被删除了。那么同样分配四个桶，把这些key安排的更紧凑一些，如右图。这就是等量扩容的意义所在。

map为什么是无序的？

因为在进行翻倍扩容之后，旧桶中的每个key要被分流到两个新桶中，在搬迁前后bucket序号可能和原来的相等，也可能比原来加上2^B。那么位置就无法确定，这样key的位置就发生了重大的变化，遍历map的结果就不可能按原来的顺序了。

但是，如果我们创建了一个map，不对它进行插入、删除、修改的操作，按理说每次遍历这样的map，都会返回一个固定顺序的key/value序列吧。

在Go1.0之前，Map确实是稳定迭代的，但是在1.0之后，GO杜绝了这种做法，因为这样会给新手程序员带来误解，以为这是一定会发生的事情，但是Go并不保证每个Go版本的迭代遍历规则都是一样的，会导致移植性问题。

因此，Go会生成一个随机数，随机从一个bucket来进行遍历，并且，还会随机从某个cell（tophash）开始遍历。

// 生成随机数 r
r := uintptr(fastrand())
if h.B > 31-bucketCntBits {
	r += uintptr(fastrand()) << 31
}

// 从哪个 bucket 开始遍历
it.startBucket = r & (uintptr(1)<<h.B - 1)
// 从 bucket 的哪个 cell 开始遍历
it.offset = uint8(r >> h.B & (bucketCnt - 1))

key/value为什么要分开来存？

这里涉及到内存对齐的概念。

在64位操作系统中，一次会读8字节，在32位操作系统中，一次会读4字节。

假如说，有这样一个类型的map：

map[int64]int8

它的key是int64，也就是8字节，value是int8，也就是1字节。

如果是key/value key/value这样的形式存储的话。

那么因为key占8字节，value占1字节，下一个紧跟着的key也占8字节，value占1字节。

因为64位操作系统一次要读8字节，所以value的7字节就空余出来了。

而如果合到一起存的话，8个key存到一起，不需要有空闲。

8个value存在一起，每一个value虽然占1字节，但是还是一次拿8字节，然后通过地址就能得到每一个字节的数据了。

map进阶问题

可以边遍历边删除吗？

map不是线程安全的。 同时读写一个map被检测到会触发panic。

可以通过sync.RWMutex读写锁来解决。

读之前调用 RLock() 函数，读完之后调用 RUnlock() 函数解锁；写之前调用 Lock() 函数，写完之后，调用 Unlock() 解锁。

另外，sync.Map 是线程安全的 map，也可以使用。它的实现原理，这次先不说了。

key可以是float类型吗？

可以，但是存的时候，会先把它转成int类型，再插入，有可能存的是近似值。