分享老師：學而思網校 郭雨田
一、map的結構與設計原理
golang中map是一個kv對集合。底層使用hash table，用連結清單來解決沖突 ，出現沖突時，不是每一個key都申請一個結構通過連結清單串起來，而是以bmap為最小粒度挂載，一個bmap可以放8個kv。在哈希函數的選擇上，會在程式啟動時，檢測 cpu 是否支援 aes，如果支援，則使用 aes hash，否則使用 memhash。每個map的底層結構是hmap，是有若幹個結構為bmap的bucket組成的數組。每個bucket底層都采用連結清單結構。接下來，我們來詳細看下map的結構：
// A header for a Go map.type hmap struct { count int // 元素個數 flags uint8 B uint8 // 擴容常量相關字段B是buckets數組的長度的對數 2^B noverflow uint16 // 溢出的bucket個數 hash0 uint32 // hash seed buckets unsafe.Pointer // buckets 數組指針 oldbuckets unsafe.Pointer // 結構擴容的時候用于指派的buckets數組 nevacuate uintptr // 搬遷進度 extra *mapextra // 用于擴容的指針}type mapextra struct { overflow *[]*bmap oldoverflow *[]*bmap nextOverflow *bmap}// A bucket for a Go map.type bmap struct { tophash [bucketCnt]uint8 // len為8的數組}//底層定義的常量 const ( // Maximum number of key/value pairs a bucket can hold. bucketCntBits = 3 bucketCnt = 1 << bucketCntBits)
但這隻是表面(src/runtime/hashmap.go)的結構，編譯期間會給它加料，動态地建立一個新的結構：
type bmap struct { topbits [8]uint8 keys [8]keytype values [8]valuetype pad uintptr overflow uintptr}
hmap和bmap的結構是這樣的 ：
bmap
就是我們常說的“桶”，桶裡面會最多裝 8 個 key，這些 key 之是以會落入同一個桶，是因為它們經過哈希計算後，哈希結果是“一類”的，關于key的定位我們在map的 查詢和指派中詳細說明。在桶内，又會根據 key 計算出來的 hash 值的高 8 位來決定 key 到底落入桶内的哪個位置(一個桶内最多有8個位置)。 當 map 的 key 和 value 都不是指針，并且 size 都小于 128 位元組的情況下，會把 bmap 标記為不含指針，這樣可以避免 gc 時掃描整個 hmap。但是，我們看 bmap 其實有一個 overflow 的字段，是指針類型的，破壞了 bmap 不含指針的設想，這時會把 overflow 移動到 hmap的extra 字段來。這部分我們在分析擴容操作的時候再詳細說明。下面我們看下bmap的内部組成圖：
HOBHash
指的就是 top hash，每個bucket中topHash唯一。key 和 value 是各自放在一起的，并不是
key/value/...
這樣的形式。可以省略掉 padding 字段，節省記憶體空間。例如，有這樣一個類型的 map：
map[int64]int8
，如果按照
key/value...
這樣的模式存儲，那在每一個 key/value 對之後都要額外 padding 7 個位元組；而将所有的 key，value 分别綁定到一起，這種形式
key/key/.../value/value/...
，則隻需要在最後添加 padding，每個 bucket 設計成最多隻能放 8 個 key-value 對，如果有第 9 個 key-value 落入目前的 bucket，那就需要再建構一個 bucket ，通過
overflow
指針連接配接起來。
二、map操作底層原理分析
1、map初始化：
方法1：var m map[string]string // 聲明變量 --nil map 支援查詢 傳回類型預設值 指派、delete操作會panicm = make(map[string]string, 10) // 初始化 --empty map 可以進行指派操作了方法2：m := make(map[string]string,10) // 容量參數可省略方法3：m := map[string]string{ // 通過直接指派進行初始化 "test": "test", "name": "lili", "age": "one", }
第一步：入參校驗，判斷key的類型是否合法，必須為可比較類型。
第二步：底層調用makemap函數，計算得到合适的B，map容量最多可容納6.5*2^B個元素，6.5為裝載因子門檻值常量。 裝載因子的計算公式是：裝載因子=填入表中的元素個數/散清單的長度，裝載因子越大，說明空閑位置越少，沖突越多，散清單的性能會下降。
func makemap(t *maptype, hint int, h *hmap) *hmap {//邊界校驗 if hint < 0 || hint > int(maxSliceCap(t.bucket.size)) { hint = 0 }// initialize Hmap if h == nil { h = new(hmap) }//生成hash種子 h.hash0 = fastrand() // find size parameter which will hold the requested # of elements B := uint8(0)//計算得到合适的B for overLoadFactor(hint, B) { B++ } h.B = B // allocate initial hash table // if B == 0, the buckets field is allocated lazily later (in mapassign) // If hint is large zeroing this memory could take a while.//申請桶空間 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}//常量loadFactorNum=13 ，loadFactorDen=2func overLoadFactor(count int, B uint8) bool { return count > bucketCnt && uintptr(count) > loadFactorNum*(bucketShift(B)/loadFactorDen)
makemap函數會通過
fastrand
建立一個随機的哈希種子，然後根據傳入的
hint
計算出需要的最小需要的桶的數量，最後再使用
makeBucketArray
建立用于儲存桶的數組，這個方法其實就是根據傳入的
B
計算出的需要建立的桶數量在記憶體中配置設定一片連續的空間用于存儲資料，在建立桶的過程中還會額外建立一些用于儲存溢出資料的桶，數量是
2^(B-4)
個。初始化完成傳回hmap指針。
2、查找操作
Go 語言中讀取 map 有兩種文法：帶 comma 和 不帶 comma。當要查詢的 key 不在 map 裡，帶 comma 的用法會傳回一個 bool 型變量提示 key 是否在 map 中；而不帶 comma 的語句則會傳回一個 value 類型的零值。如果 value 是 int 型就會傳回 0，如果 value 是 string 類型，就會傳回空字元串。
value := m["name"]fmt.Printf("value:%s", value)value, ok := m["name"] if ok { fmt.Printf("value:%s", value) }
兩種文法對應到底層兩個不同的函數，那麼在底層是如何定位到key的呢？稍後我們對函數進行源碼分析。
func mapaccess1(t *maptype, h *hmap, key unsafe.Pointer) unsafe.Pointerfunc mapaccess2(t *maptype, h *hmap, key unsafe.Pointer) (unsafe.Pointer, bool)
key的定位：
key 經過哈希計算後得到哈希值，共 64 個 bit 位(64位機，32位機就不讨論了，現在主流都是64位機)，計算它到底要落在哪個桶時，隻會用到最後 B 個 bit 位。還記得前面提到過的 B 嗎？如果 B = 5，那麼桶的數量，也就是 buckets 數組的長度是 2^5 = 32。例如，現在有一個 key 經過哈希函數計算後，得到的哈希結果是：
用最後的 5 個 bit 位，也就是
01010
，值為 10，也就是 10 号桶。這個操作實際上就是取餘操作，但是取餘開銷太大，是以代碼實作上用的位操作代替。再用哈希值的高 8 位，找到此 key 在 bucket 中的位置，這是在尋找已有的 key。最開始桶内還沒有 key，新加入的 key 會找到第一個空位放入。buckets 編号就是桶編号，當兩個不同的 key 落在同一個桶中，也就是發生了哈希沖突。沖突的解決手段是用連結清單法：在 bucket 中，從前往後找到第一個空位。這樣，在查找某個 key 時，先找到對應的桶，再去周遊 bucket 中的 key。
上圖中，假定 B = 5，是以 bucket 總數就是 2^5 = 32。首先計算出待查找 key 的哈希，使用低 5 位
00110
，找到對應的 6 号 bucket，使用高 8 位
10010111
，對應十進制 151，在 6 号 bucket 中尋找 tophash 值(HOB hash)為 151 的 key，找到了 2 号槽位，這樣整個查找過程就結束了。如果在 bucket 中沒找到，并且 overflow 不為空，還要繼續去 overflow bucket 中尋找，直到找到或是所有的 key 槽位都找遍了，包括所有的 overflow bucket。接下來我們看下底層函數源碼：
func mapaccess1(t *maptype, h *hmap, key unsafe.Pointer) unsafe.Pointer { //... // 如果 h 什麼都沒有，傳回零值 if h == nil || h.count == 0 { return unsafe.Pointer(&zeroVal[0]) } // 寫和讀沖突 if h.flags&hashWriting != 0 { throw("concurrent map read and map write") } // 不同類型 key 使用的 hash 算法在編譯期确定 alg := t.key.alg // 計算哈希值，并且加入 hash0 引入随機性 hash := alg.hash(key, uintptr(h.hash0)) // 比如 B=5，那 m 就是31，二進制是全 1 // 求 bucket num 時，将 hash 與 m 相與， // 達到 bucket num 由 hash 的低 8 位決定的效果 m := bucketMask(h.B) // b 就是 bucket 的位址 b := (*bmap)(add(h.buckets, (hash&m)*uintptr(t.bucketsize))) // oldbuckets 不為 nil，說明發生了擴容 if c := h.oldbuckets; c != nil { // 如果不是同 size 擴容(看後面擴容的内容) // 對應條件 1 的解決方案 if !h.sameSizeGrow() { // 新 bucket 數量是老的 2 倍 m >>= 1 } // 求出 key 在老的 map 中的 bucket 位置 oldb := (*bmap)(add(c, (hash&m)*uintptr(t.bucketsize))) // 如果 oldb 沒有搬遷到新的 bucket // 那就在老的 bucket 中尋找 if !evacuated(oldb) { b = oldb } } // 計算出高 8 位的 hash // 相當于右移 56 位，隻取高8位 top := tophash(hash) //開始尋找key for ; b != nil; b = b.overflow(t) { // 周遊 8 個 bucket for i := uintptr(0); i < bucketCnt; i++ { // tophash 不比對，繼續 if b.tophash[i] != top { continue } // tophash 比對，定位到 key 的位置 k := add(unsafe.Pointer(b), dataOffset+i*uintptr(t.keysize)) // key 是指針 if t.indirectkey { // 解引用 k = *((*unsafe.Pointer)(k)) } // 如果 key 相等 if alg.equal(key, k) { // 定位到 value 的位置 v := add(unsafe.Pointer(b), dataOffset+bucketCnt*uintptr(t.keysize)+i*uintptr(t.valuesize)) // value 解引用 if t.indirectvalue { v = *((*unsafe.Pointer)(v)) } return v } } } return unsafe.Pointer(&zeroVal[0])}
這裡我們再詳細分析下key/value值是如何擷取的：
// key 定位公式k :=add(unsafe.Pointer(b),dataOffset+i*uintptr(t.keysize))// value 定位公式v:= add(unsafe.Pointer(b),dataOffset+bucketCnt*uintptr(t.keysize)+i*uintptr(t.valuesize))//對于 bmap 起始位址的偏移：dataOffset = unsafe.Offsetof(struct{ b bmap v int64}{}.v)
bucket 裡 key 的起始位址就是 unsafe.Pointer(b)+dataOffset。第 i 個 key 的位址就要在此基礎上跨過 i 個 key 的大小；而我們又知道，value 的位址是在所有 key 之後，是以第 i 個 value 的位址還需要加上所有 key 的偏移。
3、指派操作
m := make(map[int32]int32) m[0] = 6666666
接下來我們将分成幾個部分去看看底層在指派的時候，進行了什麼操作：
第一階段：校驗和初始化
func mapassign(t *maptype, h *hmap, key unsafe.Pointer) unsafe.Pointer { //判斷 hmap 是否已經初始化(是否為 nil) if h == nil { panic(plainError("assignment to entry in nil map")) } //... //判斷是否并發讀寫 map，若是則抛出異常 if h.flags&hashWriting != 0 { throw("concurrent map writes") } //根據 key 的不同類型調用不同的 hash 方法計算得出 hash 值 alg := t.key.alg hash := alg.hash(key, uintptr(h.hash0)) //設定 flags 标志位，表示有一個 goroutine 正在寫入資料。因為 alg.hash 有可能出現 panic 導緻異常 h.flags |= hashWriting //判斷 buckets 是否為 nil，若是則調用 newobject 根據目前 bucket 大小進行配置設定 //初始化時沒有初始 buckets，那麼它在第一次指派時就會對 buckets 配置設定 if h.buckets == nil { h.buckets = newobject(t.bucket) // newarray(t.bucket, 1) } }
第二階段：尋找可插入位和更新既有值
//根據低八位計算得到 bucket 的記憶體位址 bucket := hash & bucketMask(h.B) //判斷是否正在擴容，若正在擴容中則先遷移再接着處理 if h.growing() { growWork(t, h, bucket) } //計算并得到 bucket 的 bmap 指針位址 b := (*bmap)(unsafe.Pointer(uintptr(h.buckets) + bucket*uintptr(t.bucketsize))) //計算 key hash 高八位用于查找 Key top := tophash(hash) var inserti *uint8 var insertk unsafe.Pointer var val unsafe.Pointer for { //疊代 buckets 中的每一個 bucket(共 8 個) for i := uintptr(0); i < bucketCnt; i++ { //對比 bucket.tophash 與 top(高八位)是否一緻 if b.tophash[i] != top { //若不一緻，判斷是否為空槽 if b.tophash[i] == empty && inserti == nil { //有兩種情況，第一種是沒有插入過。第二種是插入後被删除 inserti = &b.tophash[i] insertk = add(unsafe.Pointer(b), dataOffset+i*uintptr(t.keysize)) //把該位置辨別為可插入 tophash 位置,這裡就是第一個可以插入資料的地方 val = add(unsafe.Pointer(b), dataOffset+bucketCnt*uintptr(t.keysize)+i*uintptr(t.valuesize)) } continue } //若是比對(也就是原本已經存在)，則進行更新。最後跳出并傳回 value 的記憶體位址 k := add(unsafe.Pointer(b), dataOffset+i*uintptr(t.keysize)) if t.indirectkey { k = *((*unsafe.Pointer)(k)) } if !alg.equal(key, k) { continue } // already have a mapping for key. Update it. if t.needkeyupdate { typedmemmove(t.key, k, key) } val = add(unsafe.Pointer(b), dataOffset+bucketCnt*uintptr(t.keysize)+i*uintptr(t.valuesize)) goto done } //判斷是否疊代完畢，若是則結束疊代 buckets 并更新目前桶位置 ovf := b.overflow(t) if ovf == nil { break } b = ovf } //若滿足三個條件：觸發最大 LoadFactor 、存在過多溢出桶 overflow buckets、沒有正在進行擴容。就會進行擴容動作(以確定後續的動作) 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 { // all current buckets are full, allocate a new one. newb := h.newoverflow(t, b) inserti = &newb.tophash[0] insertk = add(unsafe.Pointer(newb), dataOffset) val = add(insertk, bucketCnt*uintptr(t.keysize)) }// store new key/value at insert position if t.indirectkey { kmem := newobject(t.key) *(*unsafe.Pointer)(insertk) = kmem insertk = kmem } if t.indirectvalue { vmem := newobject(t.elem) *(*unsafe.Pointer)(val) = vmem } typedmemmove(t.key, insertk, key) *inserti = top h.count++done ... return val
第四階段：寫入
最後傳回的是記憶體位址。是怎麼進行寫入的呢？這是因為隐藏的最後一步寫入動作(将值拷貝到指定記憶體區域)是通過底層彙編配合來完成的，在 runtime 中隻完成了絕大部分的動作。
mapassign
函數和拿到值存放的記憶體位址，再将 6666666 這個值存放進該記憶體位址中。另外我們看到
PCDATA
指令，主要是包含一些垃圾回收的資訊，由編譯器産生。
...0x0099 00153 (test.go:6) CALL runtime.mapassign_fast32(SB)0x009e 00158 (test.go:6) PCDATA $2, $20x009e 00158 (test.go:6) MOVQ 24(SP), AX0x00a3 00163 (test.go:6) PCDATA $2, $00x00a3 00163 (test.go:6) MOVL $6666666, (AX)
擴容：
關于上文中一直提到的擴容是怎麼回事呢，現在我們來具體分析下：還記得bucket中的topHash字段嗎？現在我們來補充知識點minTopHash：當一個 cell 的 tophash 值小于 minTopHash 時，标志這個 cell 的遷移狀态。因為這個狀态值是放在 tophash 數組裡，為了和正常的哈希值區分開，會給 key 計算出來的哈希值一個增量：minTopHash。這樣就能區分正常的 top hash 值和表示狀态的哈希值。下面的這幾種狀态就表征了 bucket 的情況：
// 空的 cell，也是初始時 bucket 的狀态empty = 0// 空的 cell，表示 cell 已經被遷移到新的 bucketevacuatedEmpty = 1// key,value 已經搬遷完畢，但是 key 都在新 bucket 前半部分，evacuatedX = 2// 同上，key 在後半部分evacuatedY = 3// tophash 的最小正常值minTopHash = 4
為了避免計算出的topHash與minTopHash 沖突，底層做了相關操作：
func tophash(hash uintptr) uint8 { top := uint8(hash >> (sys.PtrSize*8 - 8)) if top < minTopHash { top += minTopHash } return top}
随着向 map 中添加的 key 越來越多，key 發生碰撞的機率也越來越大。bucket 中的 8 個 cell 會被逐漸塞滿，查找、插入、删除 key 的效率也會越來越低。最理想的情況是一個 bucket 隻裝一個 key，這樣，就能達到
O(1)
的效率，但這樣空間消耗太大，用空間換時間的代價太高。Go 語言采用一個 bucket 裡裝載 8 個 key，定位到某個 bucket 後，還需要再定位到具體的 key，這實際上又用了時間換空間。當然，這樣做，要有一個度，不然所有的 key 都落在了同一個 bucket 裡，直接退化成了連結清單，各種操作的效率直接降為 O(n)，是不行的。是以，需要有一個名額來衡量前面描述的情況，這就是
裝載因子
。 Go 源碼裡這樣定義:
loadFactor := count/(2^B)
count 就是 map 的元素個數，2^B 表示 bucket 數量。 再來說觸發 map 擴容的時機：在向 map 插入新 key 的時候，會進行條件檢測，符合下面這 2 個條件，就會觸發擴容： 1、裝載因子超過門檻值，源碼裡定義的門檻值是 6.5 2、overflow 的 bucket 數量過多 通過彙編語言可以找到指派操作對應源碼中的函數是
mapassign
，對應擴容條件的源碼如下：
//觸發擴容的時機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 }// 裝載因子超過 6.5func overLoadFactor(count int, B uint8) bool { return count > bucketCnt && uintptr(count) > loadFactorNum*(bucketShift(B)/loadFactorDen)}// overflow buckets 太多func tooManyOverflowBuckets(noverflow uint16, B uint8) bool { if B > 15 { B = 15 } return noverflow >= uint16(1)<}
第 1 點：我們知道，每個 bucket 有 8 個空位，在沒有溢出，且所有的桶都裝滿了的情況下，裝載因子算出來的結果是 8。是以當裝載因子超過 6.5 時，表明很多 bucket 都快要裝滿了，查找效率和插入效率都變低了。在這個時候進行擴容是有必要的。 第 2 點：是對第 1 點的補充。就是說在裝載因子比較小的情況下，這時候 map 的查找和插入效率也很低，而第 1 點識别不出來這種情況。表面現象就是計算裝載因子的分子比較小，即 map 裡元素總數少，但是 bucket 數量多(真實配置設定的 bucket 數量多，包括大量的 overflow bucket)。 不難想像造成這種情況的原因：不停地插入、删除元素。先插入很多元素，導緻建立了很多 bucket，但是裝載因子達不到第 1 點的臨界值，未觸發擴容來緩解這種情況。之後，删除元素降低元素總數量，再插入很多元素，導緻建立很多的 overflow bucket，但就是不會觸犯第 1 點的規定，你能拿我怎麼辦？overflow bucket 數量太多，導緻 key 會很分散，查找插入效率低得吓人，是以出台第 2 點規定。這就像是一座空城，房子很多，但是住戶很少，都分散了，找起人來很困難。 對于命中條件 1，2 的限制，都會發生擴容。但是擴容的政策并不相同，畢竟兩種條件應對的場景不同。 對于條件 1，元素太多，而 bucket 數量太少，很簡單：将 B 加 1，bucket 最大數量(
2^B
)直接變成原來 bucket 數量的 2 倍。于是，就有新老 bucket 了。注意，這時候元素都在老 bucket 裡，還沒遷移到新的 bucket 來。新 bucket 隻是最大數量變為原來最大數量的 2 倍(
2^B*2
) 。 對于條件 2，其實元素沒那麼多，但是 overflow bucket 數特别多，說明很多 bucket 都沒裝滿。解決辦法就是開辟一個新 bucket 空間，将老 bucket 中的元素移動到新 bucket，使得同一個 bucket 中的 key 排列地更緊密。這樣，原來，在 overflow bucket 中的 key 可以移動到 bucket 中來。結果是節省空間，提高 bucket 使用率，map 的查找和插入效率自然就會提升。 由于 map 擴容需要将原有的 key/value 重新搬遷到新的記憶體位址，如果有大量的 key/value 需要搬遷，會非常影響性能。 是以 Go map 的擴容采取了一種稱為“漸進式”的方式，原有的 key 并不會一次性搬遷完畢，每次最多隻會搬遷 2 個 bucket。 上面說的
hashGrow()
函數實際上并沒有真正地“搬遷”，它隻是配置設定好了新的 buckets，并将老的 buckets 挂到了 oldbuckets 字段上。真正搬遷 buckets 的動作在
growWork()
函數中，而調用
growWork()
函數的動作是在 mapassign 和 mapdelete 函數中。也就是插入或修改、删除 key 的時候，都會嘗試進行搬遷 buckets 的工作。先檢查 oldbuckets 是否搬遷完畢，具體來說就是檢查 oldbuckets 是否為 nil。
func hashGrow(t *maptype, h *hmap) { // B+1 相當于是原來 2 倍的空間 bigger := uint8(1) // 對應條件 2 if !overLoadFactor(h.count+1, h.B) { // 進行等量的記憶體擴容，是以 B 不變 bigger = 0 h.flags |= sameSizeGrow } // 将老 buckets 挂到 buckets 上 oldbuckets := h.buckets // 申請新的 buckets 空間 newbuckets, nextOverflow := makeBucketArray(t, h.B+bigger, nil) //先把 h.flags 中 iterator 和 oldIterator 對應位清 0 //如果 iterator 位為 1，把它轉接到 oldIterator 位，使得 oldIterator 标志位變成1 //可以了解為buckets 現在挂到了 oldBuckets 名下了，将對應的标志位也轉接過去 flags := h.flags &^ (iterator | oldIterator) if h.flags&iterator != 0 { flags |= oldIterator } // commit the grow (atomic wrt gc) h.B += bigger h.flags = flags h.oldbuckets = oldbuckets h.buckets = newbuckets // 搬遷進度為 0 h.nevacuate = 0 // overflow buckets 數為 0 h.noverflow = 0}
幾個标志位如下：
// 可能有疊代器使用 bucketsiterator = 1// 可能有疊代器使用 oldbucketsoldIterator = 2// 有協程正在向 map 中寫入 keyhashWriting = 4// 等量擴容(對應條件 2)sameSizeGrow = 8
再來看看真正執行搬遷工作的 growWork() 函數
func growWork(t *maptype, h *hmap, bucket uintptr) { // 搬遷正在使用的舊 bucket evacuate(t, h, bucket&h.oldbucketmask()) // 再搬遷一個 bucket，以加快搬遷程序 if h.growing() { evacuate(t, h, h.nevacuate) }}func (h *hmap) growing() bool { return h.oldbuckets != nil}
搬遷過程evacuate源碼：
type evacDst struct { b *bmap // 表示bucket 移動的目标位址 i int // 指向 x,y 中 key/val 的 index k unsafe.Pointer // 指向 x，y 中的 key v unsafe.Pointer // 指向 x，y 中的 value}func evacuate(t *maptype, h *hmap, oldbucket uintptr) { // 定位老的 bucket 位址 b := (*bmap)(add(h.oldbuckets, oldbucket*uintptr(t.bucketsize))) // 計算容量 結果是 2^B，如 B = 5，結果為32 newbit := h.noldbuckets() // 如果 b 沒有被搬遷過 if !evacuated(b) { // 預設是等 size 擴容，前後 bucket 序号不變 var xy [2]evacDst // 使用 x 來進行搬遷 x := &xy[0] x.b = (*bmap)(add(h.buckets, oldbucket*uintptr(t.bucketsize))) x.k = add(unsafe.Pointer(x.b), dataOffset) x.v = add(x.k, bucketCnt*uintptr(t.keysize)) // 如果不是等 size 擴容，前後 bucket 序号有變 if !h.sameSizeGrow() { // 使用 y 來進行搬遷 y := &xy[1] // y 代表的 bucket 序号增加了 2^B y.b = (*bmap)(add(h.buckets, (oldbucket+newbit)*uintptr(t.bucketsize))) y.k = add(unsafe.Pointer(y.b), dataOffset) y.v = add(y.k, bucketCnt*uintptr(t.keysize)) } // 周遊所有的 bucket，包括 overflow buckets b 是老的 bucket 位址 for ; b != nil; b = b.overflow(t) { k := add(unsafe.Pointer(b), dataOffset) v := add(k, bucketCnt*uintptr(t.keysize)) // 周遊 bucket 中的所有 cell for i := 0; i < bucketCnt; i, k, v = i+1, add(k, uintptr(t.keysize)), add(v, uintptr(t.valuesize)) { // 目前 cell 的 top hash 值 top := b.tophash[i] // 如果 cell 為空，即沒有 key if top == empty { // 那就标志它被"搬遷"過 b.tophash[i] = evacuatedEmpty continue } // 正常不會出現這種情況 // 未被搬遷的 cell 隻可能是 empty 或是 // 正常的 top hash(大于 minTopHash) if top < minTopHash { throw("bad map state") } // 如果 key 是指針，則解引用 k2 := k if t.indirectkey { k2 = *((*unsafe.Pointer)(k2)) } var useY uint8 // 如果不是等量擴容 if !h.sameSizeGrow() { // 計算 hash 值，和 key 第一次寫入時一樣 hash := t.key.alg.hash(k2, uintptr(h.hash0)) // 如果有協程正在周遊 map 如果出現 相同的 key 值，算出來的 hash 值不同 if h.flags&iterator != 0 && !t.reflexivekey && !t.key.alg.equal(k2, k2) { // useY =1 使用位置Y useY = top & 1 top = tophash(hash) } else { // 第 B 位置 不是 0 if hash&newbit != 0 { //使用位置Y useY = 1 } } } if evacuatedX+1 != evacuatedY { throw("bad evacuatedN") } //決定key是裂變到 X 還是 Y b.tophash[i] = evacuatedX + useY // evacuatedX + 1 == evacuatedY dst := &xy[useY] // evacuation destination // 如果 xi 等于 8，說明要溢出了 if dst.i == bucketCnt { // 建立一個 bucket dst.b = h.newoverflow(t, dst.b) // xi 從 0 開始計數 dst.i = 0 //key移動的位置 dst.k = add(unsafe.Pointer(dst.b), dataOffset) //value 移動的位置 dst.v = add(dst.k, bucketCnt*uintptr(t.keysize)) } // 設定 top hash 值 dst.b.tophash[dst.i&(bucketCnt-1)] = top // mask dst.i as an optimization, to avoid a bounds check // key 是指針 if t.indirectkey { // 将原 key(是指針)複制到新位置 *(*unsafe.Pointer)(dst.k) = k2 // copy pointer } else { // 将原 key(是值)複制到新位置 typedmemmove(t.key, dst.k, k) // copy value } //value同上 if t.indirectvalue { *(*unsafe.Pointer)(dst.v) = *(*unsafe.Pointer)(v) } else { typedmemmove(t.elem, dst.v, v) } // 定位到下一個 cell dst.i++ dst.k = add(dst.k, uintptr(t.keysize)) dst.v = add(dst.v, uintptr(t.valuesize)) } } // Unlink the overflow buckets & clear key/value to help GC. // bucket搬遷完畢 如果沒有協程在使用老的 buckets，就把老 buckets 清除掉，幫助gc if h.flags&oldIterator == 0 && t.bucket.kind&kindNoPointers == 0 { b := add(h.oldbuckets, oldbucket*uintptr(t.bucketsize)) ptr := add(b, dataOffset) n := uintptr(t.bucketsize) - dataOffset memclrHasPointers(ptr, n) } } // 更新搬遷進度 if oldbucket == h.nevacuate { advanceEvacuationMark(h, t, newbit) }}
擴容後，B 增加了 1，意味着 buckets 總數是原來的 2 倍，原來 1 号的桶“裂變”到兩個桶，某個 key 在搬遷前後 bucket 序号可能和原來相等，也可能是相比原來加上 2^B(原來的 B 值)，取決于 hash 值 第 6 bit 位是 0 還是 1。原理看下圖：
4、周遊操作：
1.隻擷取key for key := range m { fmt.Println(key) }2.隻擷取value for _, value := range m { fmt.Println(value) }3.有序周遊map，擷取kv keys := []string{} for k, _ := range m { keys = append(keys, k) } // 排序 sort.Strings(keys) // 有序周遊 for _, k := range keys { fmt.Println(k, m[k]) }
了解了上面 bucket 序号的變化，我們就可以回答另一個問題了： 為什麼周遊 map 是無序的？周遊的過程，就是按順序周遊 bucket，同時按順序周遊 bucket 中的 key。搬遷後，key 的位置發生了重大的變化，有些 key 飛上高枝，有些 key 則原地不動。這樣，周遊 map 的結果就不可能按原來的順序了。當然，如果我就一個 hard code 的 map，我也不會向 map 進行插入删除的操作，按理說每次周遊這樣的 map 都會傳回一個固定順序的 key/value 序列吧。的确是這樣，但是 Go 杜絕了這種做法，因為這樣會給新手程式員帶來誤解，以為這是一定會發生的事情，在某些情況下，可能會釀成大錯。當然，Go 做得更絕，當我們在周遊 map 時，并不是固定地從 0 号 bucket 開始周遊，每次都是從一個随機值序号的 bucket 開始周遊，并且是從這個 bucket 的一個随機序号的 cell 開始周遊。這樣，即使你是一個寫死的 map，僅僅隻是周遊它，也不太可能會傳回一個固定序列的 key/value 對了。
//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)}
重點是
fastrand
的部分，是一個生成随機數的方法：它生成了随機數。用于決定從哪裡開始循環疊代。更具體的話就是根據随機數，選擇一個桶位置作為起始點進行周遊疊代是以每次重新
for range map
，你見到的結果都是不一樣的。那是因為它的起始位置根本就不固定！
...// decide where to startr := 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 stateit.bucket = it.startBucket
5、更新操作：
底層操作原理參考上文
m["age"] = "two"m["name"] = "lily"
6、删除操作：
delete(m, "name")
寫操作底層的執行函數是
mapdelete
：* func mapdelete(t *maptype, h hmap, key unsafe.Pointer)它首先會檢查 h.flags 标志，如果發現寫标位是 1，直接 panic，因為這表明有其他協程同時在進行寫操作。計算 key 的哈希，找到落入的 bucket。檢查此 map 如果正在擴容的過程中，直接觸發一次搬遷操作。删除操作同樣是兩層循環，核心還是找到 key 的具體位置。尋找過程都是類似的，在 bucket 中挨個 cell 尋找。找到對應位置後，對 key 或者 value 進行“清零”操作，将 count 值減 1，将對應位置的 tophash 值置成
Empty
。
func mapdelete(t *maptype, h *hmap, key unsafe.Pointer) { if raceenabled && h != nil { callerpc := getcallerpc() pc := funcPC(mapdelete) racewritepc(unsafe.Pointer(h), callerpc, pc) raceReadObjectPC(t.key, key, callerpc, pc) } if msanenabled && h != nil { msanread(key, t.key.size) } if h == nil || h.count == 0 { return } if h.flags&hashWriting != 0 { throw("concurrent map writes") } alg := t.key.alg hash := alg.hash(key, uintptr(h.hash0)) // Set hashWriting after calling alg.hash, since alg.hash may panic, // in which case we have not actually done a write (delete). h.flags |= hashWriting bucket := hash & bucketMask(h.B) if h.growing() { growWork(t, h, bucket) } b := (*bmap)(add(h.buckets, bucket*uintptr(t.bucketsize))) top := tophash(hash)search: for ; b != nil; b = b.overflow(t) { for i := uintptr(0); i < bucketCnt; i++ { if b.tophash[i] != top { continue } k := add(unsafe.Pointer(b), dataOffset+i*uintptr(t.keysize)) k2 := k if t.indirectkey { k2 = *((*unsafe.Pointer)(k2)) } if !alg.equal(key, k2) { continue } // Only clear key if there are pointers in it. // 對key清零 if t.indirectkey { *(*unsafe.Pointer)(k) = nil } else if t.key.kind&kindNoPointers == 0 { memclrHasPointers(k, t.key.size) } v := add(unsafe.Pointer(b), dataOffset+bucketCnt*uintptr(t.keysize)+i*uintptr(t.valuesize)) // 對value清零 if t.indirectvalue { *(*unsafe.Pointer)(v) = nil } else if t.elem.kind&kindNoPointers == 0 { memclrHasPointers(v, t.elem.size) } else { memclrNoHeapPointers(v, t.elem.size) } // 高位hash清零 b.tophash[i] = empty // 個數減一 h.count-- break search } } if h.flags&hashWriting == 0 { throw("concurrent map writes") } h.flags &^= hashWriting}
7、并發操作
map 并不是一個線程安全的資料結構。同時讀寫一個 map 是不安全的，如果被檢測到，會直接 panic。解決方法1：讀寫鎖
sync.RWMutex。
type TestMap struct { M map[int]string Lock sync.RWMutex}func main() { testMap := TestMap{} testMap.M = map[int]string{1: "lili"} go func() { i := 0 for i < 10000 { testMap.Lock.RLock() fmt.Println(i, testMap.M[1]) testMap.Lock.RUnlock() i++ } }() go func() { i := 0 for i < 10000 { testMap.Lock.Lock() testMap.M[1] = "lily" testMap.Lock.Unlock() i++ } }() for { runtime.GC() }}
解決方法2：使用golang提供的 sync.Map
func main() { m := sync.Map{} m.Store(1, 1) i := 0 go func() { for i < 1000 { m.Store(1, 1) i++ } }() go func() { for i < 1000 { m.Store(2, 2) i++ } }() go func() { for i < 1000 { fmt.Println(m.Load(1)) i++ } }() for { runtime.GC() }}
參考文獻：
【1】《深度解密Go語言之map》
【2】《解剖Go語言map底層實作》
【3】《深入了解 Go map：指派和擴容遷移》
Golang源碼系列會有持續性的文章釋出，後續會在本公衆号陸續推出，感興趣的夥伴們敬請期待呦！