Go中的三種鎖包括:互斥鎖,讀寫鎖,sync.Map的安全的鎖. ## 1、互斥鎖 Go并發程序對共享資源進行訪問控制的主要手段，由標準庫代碼包中sync中的Mutex結構體表示。 ``` //Mutex 是互斥鎖， 零值是解鎖的互斥鎖， 首次使用后不得復制互斥鎖。 type Mutex struct { state int32 sema uint32 } ``` sync.Mutex包中的類型只有兩個公開的指針方法Lock和Unlock。 ``` //Locker表示可以鎖定和解鎖的對象。 type Locker interface { Lock() Unlock() } //鎖定當前的互斥量 //如果鎖已被使用，則調用goroutine //阻塞直到互斥鎖可用。 func (m *Mutex) Lock() //對當前互斥量進行解鎖 //如果在進入解鎖時未鎖定m，則為運行時錯誤。 //鎖定的互斥鎖與特定的goroutine無關。 //允許一個goroutine鎖定Mutex然后安排另一個goroutine來解鎖它。 func (m *Mutex) Unlock() ``` 聲明一個互斥鎖： ``` var mutex sync.Mutex ``` 不像C或Java的鎖類工具，我們可能會犯一個錯誤：忘記及時解開已被鎖住的鎖，從而導致流程異常。但Go由于存在defer，所以此類問題出現的概率極低。關于defer解鎖的方式如下： ``` var mutex sync.Mutex func Write() { mutex.Lock() defer mutex.Unlock() } ``` 如果對一個已經上鎖的對象再次上鎖，那么就會導致該鎖定操作被阻塞，直到該互斥鎖回到被解鎖狀態。 ``` fpackage main import ( "fmt" "sync" "time" ) func main() { var mutex sync.Mutex fmt.Println("begin lock") mutex.Lock() fmt.Println("get locked") for i := 1; i <= 3; i++ { go func(i int) { fmt.Println("begin lock ", i) mutex.Lock() fmt.Println("get locked ", i) }(i) } time.Sleep(time.Second) fmt.Println("Unlock the lock") mutex.Unlock() fmt.Println("get unlocked") time.Sleep(time.Second) } ``` 我們在for循環之前開始加鎖，然后在每一次循環中創建一個協程，并對其加鎖，但是由于之前已經加鎖了，所以這個for循環中的加鎖會陷入阻塞直到main中的鎖被解鎖， time.Sleep(time.Second) 是為了能讓系統有足夠的時間運行for循環，輸出結果如下： ``` > go run mutex.go begin lock get locked begin lock 3 begin lock 1 begin lock 2 Unlock the lock get unlocked get locked 3 ``` 這里可以看到解鎖后，三個協程會重新搶奪互斥鎖權，最終協程3獲勝。 互斥鎖鎖定操作的逆操作并不會導致協程阻塞，但是有可能導致引發一個無法恢復的運行時的panic，比如對一個未鎖定的互斥鎖進行解鎖時就會發生panic。避免這種情況的最有效方式就是使用defer。 我們知道如果遇到panic，可以使用recover方法進行恢復，但是如果對重復解鎖互斥鎖引發的panic卻是無用的（Go 1.8及以后）。 ``` package main import ( "fmt" "sync" ) func main() { defer func() { fmt.Println("Try to recover the panic") if p := recover(); p != nil { fmt.Println("recover the panic : ", p) } }() var mutex sync.Mutex fmt.Println("begin lock") mutex.Lock() fmt.Println("get locked") fmt.Println("unlock lock") mutex.Unlock() fmt.Println("lock is unlocked") fmt.Println("unlock lock again") mutex.Unlock() } ``` 運行: ``` > go run mutex.go begin lock get locked unlock lock lock is unlocked unlock lock again fatal error: sync: unlock of unlocked mutex goroutine 1 [running]: runtime.throw(0x4bc1a8, 0x1e) /home/keke/soft/go/src/runtime/panic.go:617 +0x72 fp=0xc000084ea8 sp=0xc000084e78 pc=0x427ba2 sync.throw(0x4bc1a8, 0x1e) /home/keke/soft/go/src/runtime/panic.go:603 +0x35 fp=0xc000084ec8 sp=0xc000084ea8 pc=0x427b25 sync.(*Mutex).Unlock(0xc00001a0c8) /home/keke/soft/go/src/sync/mutex.go:184 +0xc1 fp=0xc000084ef0 sp=0xc000084ec8 pc=0x45f821 main.main() /home/keke/go/Test/mutex.go:25 +0x25f fp=0xc000084f98 sp=0xc000084ef0 pc=0x486c1f runtime.main() /home/keke/soft/go/src/runtime/proc.go:200 +0x20c fp=0xc000084fe0 sp=0xc000084f98 pc=0x4294ec runtime.goexit() /home/keke/soft/go/src/runtime/asm_amd64.s:1337 +0x1 fp=0xc000084fe8 sp=0xc000084fe0 pc=0x450ad1 exit status 2 ``` 這里試圖對重復解鎖引發的panic進行recover，但是我們發現操作失敗，雖然互斥鎖可以被多個協程共享，但還是建議將對同一個互斥鎖的加鎖解鎖操作放在同一個層次的代碼中。 ## 2、讀寫鎖 讀寫鎖是針對讀寫操作的互斥鎖，可以分別針對讀操作與寫操作進行鎖定和解鎖操作 。 讀寫鎖的訪問控制規則如下： ① 多個寫操作之間是互斥的 ② 寫操作與讀操作之間也是互斥的 ③ 多個讀操作之間不是互斥的 在這樣的控制規則下，讀寫鎖可以大大降低性能損耗。 在Go的標準庫代碼包中sync中的RWMutex結構體表示為: ``` // RWMutex是一個讀/寫互斥鎖，可以由任意數量的讀操作或單個寫操作持有。 // RWMutex的零值是未鎖定的互斥鎖。 //首次使用后，不得復制RWMutex。 //如果goroutine持有RWMutex進行讀取而另一個goroutine可能會調用Lock，那么在釋放初始讀鎖之前，goroutine不應該期望能夠獲取讀鎖定。 //特別是，這種禁止遞歸讀鎖定。 這是為了確保鎖最終變得可用; 阻止的鎖定會阻止新讀操作獲取鎖定。 type RWMutex struct { w Mutex //如果有待處理的寫操作就持有 writerSem uint32 // 寫操作等待讀操作完成的信號量 readerSem uint32 //讀操作等待寫操作完成的信號量 readerCount int32 // 待處理的讀操作數量 readerWait int32 // number of departing readers } ``` sync中的RWMutex有以下幾種方法： ``` //對讀操作的鎖定 func (rw *RWMutex) RLock() //對讀操作的解鎖 func (rw *RWMutex) RUnlock() //對寫操作的鎖定 func (rw *RWMutex) Lock() //對寫操作的解鎖 func (rw *RWMutex) Unlock() //返回一個實現了sync.Locker接口類型的值，實際上是回調rw.RLock and rw.RUnlock. func (rw *RWMutex) RLocker() Locker ``` Unlock方法會試圖喚醒所有想進行讀鎖定而被阻塞的協程，而 RUnlock方法只會在已無任何讀鎖定的情況下，試圖喚醒一個因欲進行寫鎖定而被阻塞的協程。若對一個未被寫鎖定的讀寫鎖進行寫解鎖，就會引發一個不可恢復的panic，同理對一個未被讀鎖定的讀寫鎖進行讀寫鎖也會如此。 由于讀寫鎖控制下的多個讀操作之間不是互斥的，因此對于讀解鎖更容易被忽視。對于同一個讀寫鎖，添加多少個讀鎖定，就必要有等量的讀解鎖，這樣才能其他協程有機會進行操作。 ``` package main import ( "fmt" "sync" "time" ) func main() { var rwm sync.RWMutex for i := 0; i < 5; i++ { go func(i int) { fmt.Println("try to lock read ", i) rwm.RLock() fmt.Println("get locked ", i) time.Sleep(time.Second * 2) fmt.Println("try to unlock for reading ", i) rwm.RUnlock() fmt.Println("unlocked for reading ", i) }(i) } time.Sleep(time.Millisecond * 1000) fmt.Println("try to lock for writing") rwm.Lock() fmt.Println("locked for writing") } ``` 運行: ``` > go run rwmutex.go try to lock read 0 get locked 0 try to lock read 4 get locked 4 try to lock read 3 get locked 3 try to lock read 1 get locked 1 try to lock read 2 get locked 2 try to lock for writing try to unlock for reading 0 unlocked for reading 0 try to unlock for reading 2 unlocked for reading 2 try to unlock for reading 1 unlocked for reading 1 try to unlock for reading 3 unlocked for reading 3 try to unlock for reading 4 unlocked for reading 4 locked for writing ``` 這里可以看到創建了五個協程用于對讀寫鎖的讀鎖定與讀解鎖操作。在 rwm.Lock()種會對main中協程進行寫鎖定，但是for循環中的讀解鎖尚未完成，因此會造成mian中的協程阻塞。當for循環中的讀解鎖操作都完成后就會試圖喚醒main中阻塞的協程，main中的寫鎖定才會完成。 ## 3、sync.Map安全鎖 golang中的sync.Map是并發安全的，其實也就是sync包中golang自定義的一個名叫Map的結構體。 應用示例: ``` package main import ( "sync" "fmt" ) func main() { //開箱即用 var sm sync.Map //store 方法,添加元素 sm.Store(1,"a") //Load 方法，獲得value if v,ok:=sm.Load(1);ok{ fmt.Println(v) } //LoadOrStore方法，獲取或者保存 //參數是一對key：value，如果該key存在且沒有被標記刪除則返回原先的value（不更新）和true；不存在則store，返回該value 和false if vv,ok:=sm.LoadOrStore(1,"c");ok{ fmt.Println(vv) } if vv,ok:=sm.LoadOrStore(2,"c");!ok{ fmt.Println(vv) } //遍歷該map，參數是個函數，該函數參的兩個參數是遍歷獲得的key和value，返回一個bool值，當返回false時，遍歷立刻結束。 sm.Range(func(k,v interface{})bool{ fmt.Print(k) fmt.Print(":") fmt.Print(v) fmt.Println() return true }) } ``` 運行 : ``` a a c 1:a 2:c ``` sync.Map的數據結構: ``` type Map struct { // 該鎖用來保護dirty mu Mutex // 存讀的數據，因為是atomic.value類型，只讀類型，所以它的讀是并發安全的 read atomic.Value // readOnly //包含最新的寫入的數據，并且在寫的時候，會把read 中未被刪除的數據拷貝到該dirty中，因為是普通的map存在并發安全問題，需要用到上面的mu字段。 dirty map[interface{}]*entry // 從read讀數據的時候，會將該字段+1，當等于len（dirty）的時候，會將dirty拷貝到read中（從而提升讀的性能）。 misses int } ``` read的數據結構是： ``` type readOnly struct { m map[interface{}]*entry // 如果Map.dirty的數據和m 中的數據不一樣是為true amended bool } ``` entry的數據結構： ``` type entry struct { //可見value是個指針類型，雖然read和dirty存在冗余情況（amended=false），但是由于是指針類型，存儲的空間應該不是問題 p unsafe.Pointer // *interface{} } ``` Delete 方法: ``` func (m *Map) Delete(key interface{}) { read, _ := m.read.Load().(readOnly) e, ok := read.m[key] //如果read中沒有，并且dirty中有新元素，那么就去dirty中去找 if !ok && read.amended { m.mu.Lock() //這是雙檢查（上面的if判斷和鎖不是一個原子性操作） read, _ = m.read.Load().(readOnly) e, ok = read.m[key] if !ok && read.amended { //直接刪除 delete(m.dirty, key) } m.mu.Unlock() } if ok { //如果read中存在該key，則將該value 賦值nil（采用標記的方式刪除！） e.delete() } } func (e *entry) delete() (hadValue bool) { for { p := atomic.LoadPointer(&e.p) if p == nil || p == expunged { return false } if atomic.CompareAndSwapPointer(&e.p, p, nil) { return true } } } ``` Store 方法: ``` func (m *Map) Store(key, value interface{}) { // 如果m.read存在這個key，并且沒有被標記刪除，則嘗試更新。 read, _ := m.read.Load().(readOnly) if e, ok := read.m[key]; ok && e.tryStore(&value) { return } // 如果read不存在或者已經被標記刪除 m.mu.Lock() read, _ = m.read.Load().(readOnly) if e, ok := read.m[key]; ok { //如果entry被標記expunge，則表明dirty沒有key，可添加入dirty，并更新entry if e.unexpungeLocked() { //加入dirty中 m.dirty[key] = e } //更新value值 e.storeLocked(&value) //dirty 存在該key，更新 } else if e, ok := m.dirty[key]; ok { e.storeLocked(&value) //read 和dirty都沒有，新添加一條 } else { //dirty中沒有新的數據，往dirty中增加第一個新鍵 if !read.amended { //將read中未刪除的數據加入到dirty中 m.dirtyLocked() m.read.Store(readOnly{m: read.m, amended: true}) } m.dirty[key] = newEntry(value) } m.mu.Unlock() } //將read中未刪除的數據加入到dirty中 func (m *Map) dirtyLocked() { if m.dirty != nil { return } read, _ := m.read.Load().(readOnly) m.dirty = make(map[interface{}]*entry, len(read.m)) //read如果較大的話，可能影響性能 for k, e := range read.m { //通過此次操作，dirty中的元素都是未被刪除的，可見expunge的元素不在dirty中 if !e.tryExpungeLocked() { m.dirty[k] = e } } } //判斷entry是否被標記刪除，并且將標記為nil的entry更新標記為expunge func (e *entry) tryExpungeLocked() (isExpunged bool) { p := atomic.LoadPointer(&e.p) for p == nil { // 將已經刪除標記為nil的數據標記為expunged if atomic.CompareAndSwapPointer(&e.p, nil, expunged) { return true } p = atomic.LoadPointer(&e.p) } return p == expunged } //對entry 嘗試更新 func (e *entry) tryStore(i *interface{}) bool { p := atomic.LoadPointer(&e.p) if p == expunged { return false } for { if atomic.CompareAndSwapPointer(&e.p, p, unsafe.Pointer(i)) { return true } p = atomic.LoadPointer(&e.p) if p == expunged { return false } } } //read里 將標記為expunge的更新為nil func (e *entry) unexpungeLocked() (wasExpunged bool) { return atomic.CompareAndSwapPointer(&e.p, expunged, nil) } //更新entry func (e *entry) storeLocked(i *interface{}) { atomic.StorePointer(&e.p, unsafe.Pointer(i)) } ``` 因此，每次操作先檢查read，因為read 并發安全，性能好些；read不滿足，則加鎖檢查dirty，一旦是新的鍵值，dirty會被read更新。 Load方法: Load方法是一個加載方法，查找key。 ``` func (m *Map) Load(key interface{}) (value interface{}, ok bool) { //因read只讀，線程安全，先查看是否滿足條件 read, _ := m.read.Load().(readOnly) e, ok := read.m[key] //如果read沒有，并且dirty有新數據，那從dirty中查找，由于dirty是普通map，線程不安全，這個時候用到互斥鎖了 if !ok && read.amended { m.mu.Lock() // 雙重檢查 read, _ = m.read.Load().(readOnly) e, ok = read.m[key] // 如果read中還是不存在，并且dirty中有新數據 if !ok && read.amended { e, ok = m.dirty[key] // mssLocked（）函數是性能是sync.Map 性能得以保證的重要函數，目的講有鎖的dirty數據，替換到只讀線程安全的read里 m.missLocked() } m.mu.Unlock() } if !ok { return nil, false } return e.load() } //dirty 提升至read 關鍵函數，當misses 經過多次因為load之后，大小等于len（dirty）時候，講dirty替換到read里，以此達到性能提升。 func (m *Map) missLocked() { m.misses++ if m.misses < len(m.dirty) { return } //原子操作，耗時很小 m.read.Store(readOnly{m: m.dirty}) m.dirty = nil m.misses = 0 } ``` sync.Map是通過冗余的兩個數據結構(read、dirty),實現性能的提升。為了提升性能，load、delete、store等操作盡量使用只讀的read；為了提高read的key擊中概率，采用動態調整，將dirty數據提升為read；對于數據的刪除，采用延遲標記刪除法，只有在提升dirty的時候才刪除。