偶然看見這么篇文章：一道并發(fā)和鎖的golang面試題。
雖然年代久遠，但也稍有興趣。
正好最近也看到了 sync.Map，所以想試試能不能用 sync.Map 去實現(xiàn)上述的功能。
我還在 gayhub上找到了其他人用 sync.Mutex 的實現(xiàn)方式，【點擊這里】。
歸結(jié)一下
需求是這樣的：
在一個高并發(fā)的web服務(wù)器中，要限制IP的頻繁訪問?，F(xiàn)模擬100個IP同時并發(fā)訪問服務(wù)器，每個IP要重復訪問1000次。每個IP三分鐘之內(nèi)只能訪問一次。修改以下代碼完成該過程，要求能成功輸出 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其他都重新寫嗎？（捂臉）
聰明的你已經(jīng)發(fā)現(xiàn)，這個問題關(guān)鍵就是想讓你給 Ban 加一個讀寫鎖罷了。
而且條件中的三分鐘根本無傷大雅，因為這程序壓根就活不到那天。
思路
其實，原始的思路并沒有發(fā)生改變，還是用一個 BanList 去盛放哪些暫時無法訪問的用戶 id。
然后每次訪問的時候判斷一下這個用戶是否在這個 List 中。
修改
好，那我們現(xiàn)在需要一個結(jié)構(gòu)體，因為我們會并發(fā)讀取 map，所以我們直接使用 sync.Map：
type Ban struct {
M sync.Map
}
如果你點進 sync.Map 你會發(fā)現(xiàn)他真正存儲數(shù)據(jù)的是一個atomic.Value。
一個具有原子特性的 interface{}。
同時Ban這個結(jié)構(gòu)提還會有一個 IsIn 的方法用來判斷用戶 id 是否在Map中。
func (b *Ban) IsIn(user string) bool {
fmt.Printf("%s 進來了\n", user)
// Load 方法返回兩個值，一個是如果能拿到的 key 的 value
// 還有一個是否能夠拿到這個值的 bool 結(jié)果
v, ok := b.M.Load(user) // sync.Map.Load 去查詢對應(yīng) key 的值
if !ok {
// 如果沒有，說明可以訪問
fmt.Printf("名單里沒有 %s，可以訪問\n", user)
// 將用戶名存入到 Ban List 中
b.M.Store(ip, time.Now())
return false
}
// 如果有，則判斷用戶的時間距離現(xiàn)在是否已經(jīng)超過了 180 秒，也就是3分鐘
if time.Now().Second() - v.(time.Time).Second() > 180 {
// 超過則可以繼續(xù)訪問
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
}
下面看看測試的函數(shù)：
func main() {
var success int64 = 0
ban := new(Ban)
wg := sync.WaitGroup{} // 保證程序運行完成
for i := 0; i < 2; i++ { // 我們大循環(huán)兩次，每個user 連續(xù)訪問兩次
for j := 0; j < 10; j++ { // 人數(shù)預先設(shè)定為 10 個人
wg.Add(1)
go func(c int) {
defer wg.Done()
ip := fmt.Sprintf("%d", c)
if !ban.IsIn(ip) {
// 原子操作增加計數(shù)器，用來統(tǒng)計我們?nèi)藬?shù)的
atomic.AddInt64(&success, 1)
}
}(j)
}
}
wg.Wait()
fmt.Println("此次訪問量：", success)
}
其實測試的函數(shù)并沒有做大的改動，只不過，因為我們是并發(fā)去運行的，需要增加一個 sync.WaitGroup() 保證程序完整運行完畢后才退出。
我特地把運行數(shù)值調(diào)小一點，以方便測試。
把1000次請求，改為2次。100人改為10人。
所以整個代碼應(yīng)該是這樣的：
package main
import (
"fmt"
"sync"
"sync/atomic"
"time"
)
type Ban struct {
M sync.Map
}
func (b *Ban) IsIn(user string) bool {
...
}
func main() {
...
}
運行一下...
誒，似乎不太對哦，發(fā)現(xiàn)會出現(xiàn) 10~15 次不等的訪問量結(jié)果。為什么呢？
尋思著，其實因為并發(fā)導致的，看到這里了嗎？
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
}
...
}
并發(fā)發(fā)起的 sync.Map.Load 其實并沒有與 sync.Map.Store 連接起來形成原子操作。
所以如果有3個 user 同時進來，程序同時查詢，三個返回結(jié)果都會是 false（不在名單里）。
所以也就增加了訪問的數(shù)量。
其實 sync.Map 也已經(jīng)考慮到了這種情況，所以他會有一個 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
}
...
}
然后我們再運行一下，運行幾次。
發(fā)覺不會再出現(xiàn) 此次訪問量大于 10 的情況了。
深究一下
到此為止，這個場景下的代碼實現(xiàn)我們算是成功了。
但是真正限制用戶訪問的場景需求可不能這么玩，一般還是配合內(nèi)存數(shù)據(jù)庫去實現(xiàn)。
那么，如果你只想了解 sync.Map 的應(yīng)用，就到這里為止了。
然而好奇心驅(qū)使我看看 sync.Map 的實現(xiàn)，我們繼續(xù)吧。
制造問題
如果硬是要并發(fā)讀寫一個 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")
// 設(shè)置 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 有個設(shè)置某個 Key 的方法
func (a *A)SetMap(key string, value int) {
a[key] = value
fmt.Printf("Set Key %s: %d",key, a[key]) // 同協(xié)程的讀寫不會出問題
}
誒，看上去不錯，我們給 map A 類型定義了 get， set 方法，如果 golang 不允許并發(fā)讀寫 map 的話，應(yīng)該會報錯吧，我們跑一下。
> Get Key one: 1
> Set Key two: 2
喵喵喵???
為什么正常輸出了？
說好的并發(fā)讀寫報錯呢？
好吧，其實原因是上面的 map 讀寫，雖然我們設(shè)置了協(xié)程，但是對于計算機來說還是有時間差的。只要一個微小的先后，就不會造成 map 數(shù)據(jù)的讀寫異常，所以我們改一下。
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)
}
為了讓讀寫能夠盡可能碰撞，我們增加了循環(huán)。
現(xiàn)在我們可以看到了：
> fatal error: concurrent map read and map write
*這里之所以為有 panic 是因為在 map 實現(xiàn)中進行了并發(fā)讀寫的檢查。
解決問題
其實上面的例子和 go 對 sync.Mutex 鎖的入門教程很像。
我們證實了 map 并發(fā)讀寫的問題，現(xiàn)在我們嘗試來解決。
既然是讀寫造成的沖突，那我們首先考慮的便是加鎖。
我們給讀取一個鎖，寫入一個鎖。那么我們現(xiàn)在需要講單純的 A map 轉(zhuǎn)換成一個帶有鎖的結(jié)構(gòu)體：
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 都不行。
我們再跑一下代碼，發(fā)現(xiàn)可以了，不會報錯。
到此為止了嗎？
我們算是用最簡單的方法解決了眼前的問題，但是這樣真的沒問題嗎？
細心的你會發(fā)現(xiàn)，讀寫我們都加了鎖，而且沒有任何特殊條件限制，所以當我們要多次讀取 map 中的數(shù)據(jù)的時候，他喵的都會阻塞！就算我壓根不想改 map 中的 value...
盡管現(xiàn)在感覺不出來慢，但這對密集讀取來說是一個性能坑。
為了避免不必要的鎖，我們似乎還要讓代碼“聰明些”。
讀寫分離
沒錯，讀寫分離就是一個十分適用的設(shè)計思路。
我們準備一個 Read map，一個 Write map。
但這里的讀寫分離和我們平時說的不太一樣（記住我們的場景永遠是并發(fā)讀寫），我們不能實時或者定時讓寫入的 map 去同步（增刪改）到讀取的 map 中，
因為...這樣和上面的 map 操作沒有任何區(qū)別，因為讀取 map 沒有鎖，還是會發(fā)生并發(fā)沖突。
我們要解決的是，不“顯示”增刪改 map 中的 key 對應(yīng)的 value。
我們把問題再分類一下：
1. 修改（刪除）已有的 key 的 value
2. 增加不存在的 key 和 value
第一個問題：我們把 key 的 value 變成指針怎么樣？
相同的 key 指向同一個指針地址，指針地址又指向真正值的地址。
key -> &地址 -> &真正的值地址
Read 和 Write map 的值同時指向一個&地址，不論誰改，大家都會變。
當我們需要修改已有的 key 對應(yīng)的 value 時，我們修改的是&真正的值地址的值，并不會修改 key 對應(yīng)的&地址或值。
同時，通過atomic包，我們能夠做到對指針修改的原子性。
太棒了，修改已有的 key 問題解決。
第二個問題：因為并不存在這個 key，所以我們一定會增加新 key，
既然我們有了 Read map & Write map，那我們可以利用起來呀，
我們在 Write map 中加鎖并增加這個 key 和對應(yīng)的值，這樣不影響 Read map 的讀取。
不過，Read map 我們終究是要更新的，所以我們加一個計數(shù)器 misses，到了一定條件，我們把 Write map 安全地同步到 Read map 中，并且清空 Write map。
Read map 可以看做是 Write map 的一個只讀拷貝，不允許自行增加新 key，只能讀或者改。
上面的思想其實和 sync.Map 的實現(xiàn)離得很近了。
只不過，sync.Map 把我們的 Write map 叫做了 dirty，把 Write map 同步到 Read map 叫做了 promote（升級）。
又增加了一些結(jié)構(gòu)體封裝，和狀態(tài)標識。
其實 google 一下你就會發(fā)現(xiàn)很多分析 sync.Map 源碼的文章，都寫得很好。我這里也不贅述了，但是我想用我的理解去概括下
sync.Map 中的方法思路。
結(jié)合 sync.Map 源碼食用味道更佳。
讀取 Map.Load
1. Read map 直接讀得到嗎？
2. 么有？好吧，我們上鎖，再讀一次 Read map
3. 還沒有？那我只能去讀 Dirty map 了
4. 讀到了，不錯，我們記錄下這次讀取屬于未命中（misses + 1），順便看看我們的 dirty 是不是可以升級成 Read 了
5. 解鎖
*這里2中之所以再上鎖，是為了double-checking，防止在極小的時間差內(nèi)產(chǎn)生臟讀（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
編不下去了
大致就是這樣的思路，我這里再推薦一些正統(tǒng)的源碼分析和基準測試，相信看完以后會對 sync.Map 更加清晰。
另外，如果你注意到 Map.Store 中第6步的全部復制的話，你就會有預感，sync.Map 的使用場景其實不太適合高并發(fā)寫的邏輯。
的確，官方說明也明確指出了 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 只是幫助優(yōu)化了兩個使用場景：
1. 多讀少寫
2. 多 goroutine 操作鍵值
其實 sync.Map 還是在性能和安全之間，找了一個自己覺得合適的平衡點，就如同我們開頭的案例一樣，其實 sync.Map 也并不適用。
另外，這里有一個 【sync.Map 的進階版本】。
*atomic 和 mutex
其實在很久以前翻看 sync Map 源碼的時候，我不經(jīng)會拋出疑問，如果能夠用 atomic 來解決并發(fā)安全問題，為什么還要 mutex 呢？
而且，在進行 map.Store 的過程中，還是會直接修改 read 的 key 所對應(yīng)的值（并且無鎖狀態(tài)），這和普通修改一個 key 的值有什么區(qū)別呢？
如果 atomic 可以保證原子性，那和 mutex 有什么區(qū)別呢？
在翻查了一些資料后，我知道了：
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.
互斥鎖其實是通過阻塞其他協(xié)程起到了原子操作的功能，但是 atomic 是通過控制更底層的 CPU 指令，來達到值操作的原子性的。
所以 atomic 和 mutex 并不是一個層面的東西，而且在專職點上也不盡相同，mutex 很多地方也是通過 atomic 去實現(xiàn)的。
而 sync Map 很巧妙地將兩個結(jié)合來實現(xiàn)并發(fā)安全。
1. 它用一個指針來存儲 key 對應(yīng)的 value，當要修改的時候只是修改 value 的地址（并且是地址值的副本操作），這個可以通過 atomic 的 Pointer 操作來實現(xiàn)，并且不會又沖突。
2. 另外，又使用了讀寫分離+mutex互斥鎖，來控制 增刪改查 key 的操作，防止沖突。
其中第一點是很多源碼解讀中常常一筆帶過的，然而萌新我覺得反而是相當重要的技巧（捂臉）。
*一些疑問
misses 條件
一直沒有明白，為什么從 dirty map 升級成 read map 的條件是
misses 次數(shù)大于等于 len(m.dirty)？
Go map 為什么不加鎖？
我們可以看到下面兩篇關(guān)于不加鎖的敘述：
1. https://golang.org/doc/faq#atomic_maps
2. https://blog.golang.org/go-maps-in-action
