Go sync.Map 看一看

偶然看見這麼篇文章：一道併發和鎖的golang面試題。
雖然年代久遠，但也稍有興趣。

正好最近也看到了 sync.Map，因此想試試能不能用 sync.Map 去實現上述的功能。

我還在 gayhub上找到了其餘人用 sync.Mutex 的實現方式，【點擊這裏】。

歸結一下

需求是這樣的：

在一個高併發的web服務器中，要限制IP的頻繁訪問。現模擬100個IP同時併發訪問服務器，每一個IP要重複訪問1000次。每一個IP三分鐘以內只能訪問一次。修改如下代碼完成該過程，要求能成功輸出 success: 100。

而且給出了原始代碼：

package main

import (
    "fmt"
    "time"
)

type Ban struct {
    visitIPs map[string]time.Time
}

func NewBan() *Ban {
    return &Ban{visitIPs: make(map[string]time.Time)}
}

func (o *Ban) visit(ip string) bool {
    if _, ok := o.visitIPs[ip]; ok {
        return true
    }
    o.visitIPs[ip] = time.Now()
    return false
}

func main() {
    success := 0
    ban := NewBan()
    for i := 0; i < 1000; i++ {
        for j := 0; j < 100; j++ {
            go func() {
                ip := fmt.Sprintf("192.168.1.%d", j)
                if !ban.visit(ip) {
                    success++
                }
            }()
        }
    }
    fmt.Println("success: ", success)
}

哦吼，看到源代碼我想說，我能只留個package main其餘都從新寫嗎？（捂臉）

聰明的你已經發現，這個問題關鍵就是想讓你給 Ban 加一個讀寫鎖罷了。
並且條件中的三分鐘根本無傷大雅，由於這程序壓根就活不到那天。

思路

其實，原始的思路並無發生改變，仍是用一個 BanList 去盛放哪些暫時沒法訪問的用戶 id。
而後每次訪問的時候判斷一下這個用戶是否在這個 List 中。

修改

好，那咱們如今須要一個結構體，由於咱們會併發讀取 map，因此咱們直接使用 sync.Map：

type Ban struct {
    M sync.Map
}

若是你點進 sync.Map 你會發現他真正存儲數據的是一個atomic.Value。
一個具備原子特性的 interface{}。

同時Ban這個結構提還會有一個 IsIn 的方法用來判斷用戶 id 是否在Map中。

func (b *Ban) IsIn(user string) bool {
    fmt.Printf("%s 進來了\n", user)
    // Load 方法返回兩個值，一個是若是能拿到的 key 的 value
    // 還有一個是否可以拿到這個值的 bool 結果
    v, ok := b.M.Load(user) // sync.Map.Load 去查詢對應 key 的值
    if !ok {
        // 若是沒有，說明能夠訪問
        fmt.Printf("名單裏沒有 %s，能夠訪問\n", user)
        // 將用戶名存入到 Ban List 中
        b.M.Store(ip, time.Now())
        return false
    }
    // 若是有，則判斷用戶的時間距離如今是否已經超過了 180 秒，也就是3分鐘
    if time.Now().Second() - v.(time.Time).Second() > 180 {
        // 超過則能夠繼續訪問
        fmt.Printf("時間爲：%d-%d\n", v.(time.Time).Second(), time.Now().Second())
        // 同時從新存入時間
        b.M.Store(ip, time.Now())
        return false
    }
    // 不然不能訪問
    fmt.Printf("名單裏有 %s，拒絕訪問\n", user)
    return true
}

下面看看測試的函數：

func main() {
    var success int64 = 0
    ban := new(Ban)
    wg := sync.WaitGroup{} // 保證程序運行完成
    for i := 0; i < 2; i++ { // 咱們大循環兩次，每一個user 連續訪問兩次
        for j := 0; j < 10; j++ { // 人數預先設定爲 10 我的
            wg.Add(1)
            go func(c int) {
                defer wg.Done()
                ip := fmt.Sprintf("%d", c)
                if !ban.IsIn(ip) {
                    // 原子操做增長計數器，用來統計咱們人數的
                    atomic.AddInt64(&success, 1)
                }
            }(j)
        }
    }
    wg.Wait()
    fmt.Println("這次訪問量：", success)
}

其實測試的函數並無作大的改動，只不過，由於咱們是併發去運行的，須要增長一個 sync.WaitGroup() 保證程序完整運行完畢後才退出。

我特意把運行數值調小一點，以方便測試。
把1000次請求，改成2次。100人改成10人。

因此整個代碼應該是這樣的：

package main

import (
    "fmt"
    "sync"
    "sync/atomic"
    "time"
)

type Ban struct {
    M sync.Map
}

func (b *Ban) IsIn(user string) bool {
    ...
}

func main() {
    ...
}

運行一下...

誒，彷佛不太對哦，發現會出現 10~15 次不等的訪問量結果。爲何呢？
尋思着，其實由於併發致使的，看到這裏了嗎？

func (b *Ban) IsIn(user string) bool {
    ...
    v, ok := b.M.Load(user)
    if !ok {
        fmt.Printf("名單裏沒有 %s，能夠訪問\n", user)
        b.M.Store(ip, time.Now())
        return false
    }
    ...
}

併發發起的 sync.Map.Load 其實並無與 sync.Map.Store 鏈接起來造成原子操做。
因此若是有3個 user 同時進來，程序同時查詢，三個返回結果都會是 false（不在名單裏）。
因此也就增長了訪問的數量。

其實 sync.Map 也已經考慮到了這種狀況，因此他會有一個 LoadOrStore 的原子方法--
若是 Load 不出，就直接 Store，若是 Load 出來，那啥也不作。

因此咱們小改一下 IsIn 的代碼：

func (b *Ban) IsIn(user string) bool {
    ...
    v, ok := b.M.LoadOrStore(user, time.Now())
    if !ok {
        fmt.Printf("名單裏沒有 %s，能夠訪問\n", user)
        // 刪除b.M.Store(ip, time.Now())
        return false
    }
    ...
}

而後咱們再運行一下，運行幾回。
發覺不會再出現 這次訪問量大於 10 的狀況了。

深究一下

到此爲止，這個場景下的代碼實現咱們算是成功了。
可是真正限制用戶訪問的場景需求可不能這麼玩，通常仍是配合內存數據庫去實現。

那麼，若是你只想瞭解 sync.Map 的應用，就到這裏爲止了。
然而好奇心驅使我看看 sync.Map 的實現，咱們繼續吧。

製造問題

若是硬是要併發讀寫一個 go map 會怎麼樣？
試一下：

先來個主角 A

type A map[string]int

咱們定義成了本身一個類型 A，他骨子裏仍是 map。

type A map[string]int

func main() {
    // 初始化一個 A
    m := make(A)
    m.SetMap("one", 1)
    m.SetMap("two", 3)

    // 讀取 one
    go m.ReadMap("one")
    // 設置 two 值爲 2
    go m.SetMap("two", 2)

    time.Sleep(1*time.Second)
}

// A 有個讀取某個 Key 的方法
func (a *A)ReadMap(key string) {
    fmt.Printf("Get Key %s: %d",key, a[key])
}
// A 有個設置某個 Key 的方法
func (a *A)SetMap(key string, value int) {
    a[key] = value
    fmt.Printf("Set Key %s: %d",key, a[key]) // 同協程的讀寫不會出問題
}

誒，看上去不錯，咱們給 map A 類型定義了 get， set 方法，若是 golang 不容許併發讀寫 map 的話，應該會報錯吧，咱們跑一下。

> Get Key one: 1
> Set Key two: 2

喵喵喵???
爲何正常輸出了？
說好的併發讀寫報錯呢？
好吧，其實緣由是上面的 map 讀寫，雖然咱們設置了協程，可是對於計算機來講仍是有時間差的。只要一個微小的前後，就不會形成 map 數據的讀寫異常，因此咱們改一下。

func main() {
    m := make(A)
    m["one"] = 1
    m["two"] = 3

    go func() {
        for {
            m.ReadMap("one")
        }
    }()

    go func(){
        for {
            m.SetMap("two", 2)
        }
    }()

    time.Sleep(1*time.Second)
}

爲了讓讀寫可以儘量碰撞，咱們增長了循環。
如今咱們能夠看到了：

> fatal error: concurrent map read and map write

*這裏之因此爲有 panic 是由於在 map 實現中進行了併發讀寫的檢查。

解決問題

其實上面的例子和 go 對 sync.Mutex 鎖的入門教程很像。
咱們證明了 map 併發讀寫的問題，如今咱們嘗試來解決。

既然是讀寫形成的衝突，那咱們首先考慮的即是加鎖。
咱們給讀取一個鎖，寫入一個鎖。那麼咱們如今須要講單純的 A map 轉換成一個帶有鎖的結構體：

type A struct {
    Value map[string]int
    mu sync.Mutex
}

Value 成了真正存放咱們值的地方。
咱們還要修改下 ReadMap 和 SetMap 兩個方法。

func (a *A)ReadMap(key string) {
    a.mu.Lock()
    fmt.Printf("Get Key %s: %d",key, a.Value[key])
    a.mu.Unlock()
}
func (a *A)SetMap(key string, value int) {
    a.mu.Lock()
    a.Value[key] = value
    a.mu.Unlock()
    fmt.Printf("Set Key %s: %d",key, a.Value[key])
}

注意，這裏兩個方法中，哪個少了 Lock 和 Unlock 都不行。

咱們再跑一下代碼，發現能夠了，不會報錯。

到此爲止了嗎？

咱們算是用最簡單的方法解決了眼前的問題，可是這樣真的沒問題嗎？
細心的你會發現，讀寫咱們都加了鎖，並且沒有任何特殊條件限制，因此當咱們要屢次讀取 map 中的數據的時候，他喵的都會阻塞！就算我壓根不想改 map 中的 value...
儘管如今感受不出來慢，但這對密集讀取來講是一個性能坑。

爲了不沒必要要的鎖，咱們彷佛還要讓代碼「聰明些」。

讀寫分離

沒錯，讀寫分離就是一個十分適用的設計思路。
咱們準備一個 Read map，一個 Write map。

但這裏的讀寫分離和咱們平時說的不太同樣（記住咱們的場景永遠是併發讀寫），咱們不能實時或者定時讓寫入的 map 去同步（增刪改）到讀取的 map 中，
由於...這樣和上面的 map 操做沒有任何區別，由於讀取 map 沒有鎖，仍是會發生併發衝突。

咱們要解決的是，不「顯式」增刪改 map 中的 key 對應的 value。
咱們把問題再分類一下：

1. 修改（刪除）已有的 key 的 value
2. 增長不存在的 key 和 value

第一個問題：咱們把 key 的 value 變成指針怎麼樣？
相同的 key 指向同一個指針地址，指針地址又指向真正值的地址。

key -> &地址 -> &真正的值地址

Read 和 Write map 的值同時指向一個&地址，不論誰改，你們都會變。
當咱們須要修改已有的 key 對應的 value 時，咱們修改的是&真正的值地址的值，並不會修改 key 對應的&地址或值。
同時，經過atomic包，咱們可以作到對指針修改的原子性。
太棒了，修改已有的 key 問題解決。

第二個問題：由於並不存在這個 key，因此咱們必定會增長新 key，
既然咱們有了 Read map & Write map，那咱們能夠利用起來呀，
咱們在 Write map 中加鎖並增長這個 key 和對應的值，這樣不影響 Read map 的讀取。

不過，Read map 咱們終究是要更新的，因此咱們加一個計數器 misses，到了必定條件，咱們把 Write map 安全地同步到 Read map 中，而且清空 Write map。

Read map 能夠看作是 Write map 的一個只讀拷貝，不容許自行增長新 key，只能讀或者改。

上面的思想其實和 sync.Map 的實現離得很近了。
只不過，sync.Map 把咱們的 Write map 叫作了 dirty，把 Write map 同步到 Read map 叫作了 promote（升級）。
又增長了一些結構體封裝，和狀態標識。

其實 google 一下你就會發現不少分析 sync.Map 源碼的文章，都寫得很好。我這裏也不贅述了，可是我想用個人理解去歸納下
sync.Map 中的方法思路。

結合 sync.Map 源碼食用味道更佳。

讀取 Map.Load

1. Read map 直接讀獲得嗎？
2. 麼有？好吧，咱們上鎖，再讀一次 Read map
3. 尚未？那我只能去讀 Dirty map 了
4. 讀到了，不錯，咱們記錄下此次讀取屬於未命中（misses + 1），順便看看咱們的 dirty 是否是能夠升級成 Read 了
5. 解鎖

*這裏2中之因此再上鎖，是爲了double-checking，防止在極小的時間差內產生髒讀（dirty忽然升級 Read）。

寫入 Map.Store

1. Read map 有沒有這個 key ？
2. 有，那咱們原子操做直接修改值指針唄
3. 沒有？依舊上鎖再看看有沒有？
4. 尚未，好吧，看看 Dirty map
5. 有誒！那就修改 Dirty map 這個 key 的值指針
6. 沒有？那就要在 Dirty map 新增一個 key 咯，爲了方便以後 Dirty map 升級成 Read map，咱們還要把原先的 Read map 全複製過來
7. 解鎖

刪除 Map.Delete

1. Read map 有這個 key 嗎？
2. 有啊，那就把 value 直接改爲 nil（防止以後讀取沒有 key 還要去加鎖，影響性能）
3. 沒有？直接刪 dirty 裏的這個 key 吧

讀取或者存 Map.LoadOrStore

emmmm......

• Map.Load + Map.Store

編不下去了

大體就是這樣的思路，我這裏再推薦一些正統的源碼分析和基準測試，相信看完之後會對 sync.Map 更加清晰。

• 不得不知道的Golang之sync.Map源碼分析
• Go sync.Map 實現
• Go 1.9 sync.Map揭祕

另外，若是你注意到 Map.Store 中第6步的所有複製的話，你就會有預感，sync.Map 的使用場景其實不太適合高併發寫的邏輯。
的確，官方說明也明確指出了 sync.Map 適用的場景：

// Map is like a Go map[interface{}]interface{} but is safe for concurrent use
// by multiple goroutines without additional locking or coordination.
...
// The Map type is specialized. Most code should use a plain Go map instead,
// with separate locking or coordination, for better type safety and to make it
// easier to maintain other invariants along with the map content.
//
// The Map type is optimized for two common use cases: (1) when the entry for a given
// key is only ever written once but read many times, as in caches that only grow,
// or (2) when multiple goroutines read, write, and overwrite entries for disjoint
// sets of keys. In these two cases, use of a Map may significantly reduce lock
// contention compared to a Go map paired with a separate Mutex or RWMutex.

sync.Map 只是幫助優化了兩個使用場景：

1. 多讀少寫
2. 多 goroutine 操做鍵值

其實 sync.Map 仍是在性能和安全之間，找了一個本身以爲合適的平衡點，就如同咱們開頭的案例同樣，其實 sync.Map 也並不適用。
另外，這裏有一個 【sync.Map 的進階版本】。

---

*atomic 和 mutex

其實在好久之前翻看 sync Map 源碼的時候，我不經會拋出疑問，若是可以用 atomic 來解決併發安全問題，爲何還要 mutex 呢？
並且，在進行 map.Store 的過程當中，仍是會直接修改 read 的 key 所對應的值（而且無鎖狀態），這和普通修改一個 key 的值有什麼區別呢？

若是 atomic 能夠保證原子性，那和 mutex 有什麼區別呢？
在翻查了一些資料後，我知道了：

Mutexes are slow, due to the setup and teardown, and due to the fact that they block other goroutines for the duration of the lock.

Atomic operations are fast because they use an atomic CPU instruction, rather than relying on external locks to.

互斥鎖實際上是經過阻塞其餘協程起到了原子操做的功能，可是 atomic 是經過控制更底層的 CPU 指令，來達到值操做的原子性的。

因此 atomic 和 mutex 並非一個層面的東西，並且在專職點上也不盡相同，mutex 不少地方也是經過 atomic 去實現的。

而 sync Map 很巧妙地將兩個結合來實現併發安全。

1. 它用一個指針來存儲 key 對應的 value，當要修改的時候只是修改 value 的地址（而且是地址值的副本操做），這個能夠經過 atomic 的 Pointer 操做來實現，而且不會又衝突。
2. 另外，又使用了讀寫分離+mutex互斥鎖，來控制 增刪改查 key 的操做，防止衝突。

其中第一點是不少源碼解讀中經常一筆帶過的，然而萌新我以爲反而是至關重要的技巧（捂臉）。

*一些疑問

misses 條件

一直沒有明白，爲何從 dirty map 升級成 read map 的條件是
misses 次數大於等於 len(m.dirty)？

Go map 爲何不加鎖？

咱們能夠看到下面兩篇關於不加鎖的敘述：

1. https://golang.org/doc/faq#at...
2. https://blog.golang.org/go-ma...

*參考

• https://www.jianshu.com/p/aa0...
• https://www.zhihu.com/questio...
• https://juejin.im/post/5ae014...
• https://juejin.im/post/5b1b3d...
• http://wudaijun.com/2018/02/g...
• https://colobu.com/2017/07/11...