我們可以利用互斥鎖來保護(hù)代碼中的關(guān)鍵部分，從而確保每次只能有一個 goroutine 訪問共享資源。這樣一來，就能避免競爭條件的問題。幾乎所有支持并發(fā)編程的語言中，都使用了類似互斥鎖的機(jī)制。在本章中，我們首先會了解互斥鎖的功能。之后，還會探討一種名為“讀寫互斥鎖”的特殊類型的互斥鎖。

讀寫互斥鎖能在我們只需要在修改共享資源時才阻止并發(fā)操作的情況下，幫助我們提升性能。利用這種機(jī)制，我們可以允許多個進(jìn)程同時讀取共享資源，而與此同時，又能確保只有某個進(jìn)程能夠進(jìn)行寫入操作。我們將通過一個示例來了解讀寫互斥鎖的用法，還會學(xué)習(xí)其內(nèi)部原理，并親手實現(xiàn)一個類似的互斥鎖。

使用互斥量保護(hù)臨界區(qū)

如果我們能確保每次只有一條執(zhí)行線程能夠訪問那些關(guān)鍵代碼段，那該多好啊。這就是互斥鎖的作用所在。可以把互斥鎖看作是一種“物理鎖”，它能防止多個協(xié)程同時訪問代碼中的特定部分。只要每次只有一個協(xié)程在訪問關(guān)鍵代碼段，就能避免競爭條件的問題。因為，競爭條件只有在那兩個或多個協(xié)程同時嘗試訪問同一資源時才會發(fā)生。

我們可以利用互斥鎖來標(biāo)記代碼中臨界區(qū)的起始和結(jié)束位置。當(dāng)某個協(xié)程進(jìn)入由互斥鎖保護(hù)的臨界區(qū)時，它首先會通過程序代碼中的指令明確地鎖定該互斥鎖。之后，協(xié)程開始執(zhí)行臨界區(qū)內(nèi)的代碼。當(dāng)執(zhí)行完成后，它會釋放互斥鎖，這樣其他協(xié)程就可以進(jìn)入該臨界區(qū)了。

如果另一個協(xié)程試圖鎖定一個已經(jīng)被占用的互斥鎖，那么該協(xié)程將被掛起，直到互斥鎖被釋放為止。如果有多個協(xié)程都在等待鎖的釋放，那么只會有一個協(xié)程能夠繼續(xù)執(zhí)行，而那個協(xié)程就是下一個成功獲得互斥鎖的協(xié)程。

所謂“互斥鎖”，其實就是一種用于防止競爭條件的并發(fā)控制機(jī)制。它確保同一時間只能有一個執(zhí)行單元（比如一個協(xié)程或內(nèi)核級線程）進(jìn)入臨界區(qū)域。如果有兩個執(zhí)行單元同時試圖獲取對該互斥鎖的訪問權(quán)，那么互斥鎖的規(guī)則會保證只有一個協(xié)程能夠成功獲取訪問權(quán)，而另一個執(zhí)行單元則必須等待，直到互斥鎖再次變?yōu)榭捎脿顟B(tài)。

在 Go 語言中，互斥鎖的功能由 `sync` 包中的 `Mutex` 類提供。通過 `Mutex` 類，我們可以使用 `Lock()` 和 `Unlock()` 這兩個方法來實現(xiàn)對資源的鎖定與解鎖操作。

package main

import (
  "fmt"
  "sync"
  "time"
)

func stingy(money *int, mutex *sync.Mutex) {
  for i := 0; i < 1000000; i++ {
    // 進(jìn)入臨界區(qū)之前，先鎖定互斥變量
    mutex.Lock()
    *money += 10
    // 離開臨界區(qū)之后，解鎖互斥變量
    mutex.Unlock()
  }
  fmt.Println("Stingy Done")
}

func spendy(money *int, mutex *sync.Mutex) {
  for i := 0; i < 1000000; i++ {
    mutex.Lock()
    *money -= 10
    mutex.Unlock()
  }
  fmt.Println("Spendy Done")
}

func main() {
  money := 100
  mutex := sync.Mutex{}
  go stingy(&money, &mutex)
  go spendy(&money, &mutex)
  time.Sleep(2 * time.Second)
  mutex.Lock()
  fmt.Println("Money in bank account: ", money)
  mutex.Unlock()
}

在我們的主函數(shù)中，當(dāng)各個協(xié)程執(zhí)行完成后，我們在讀取“money”變量時也會使用互斥鎖來確保同步。雖然由于我們設(shè)置了等待時間來確保所有協(xié)程都已完成執(zhí)行，因此出現(xiàn)競爭條件的概率很低。但即便如此，保護(hù)共享資源始終是良好的編程習(xí)慣。使用互斥鎖（以及后面章節(jié)中介紹的其他同步機(jī)制），可以確保協(xié)程能夠讀取到該變量的最新值。

請注意：我們必須保護(hù)所有關(guān)鍵代碼段，包括那些 goroutine 僅用于讀取共享資源的區(qū)域。編譯器的優(yōu)化措施可能會重新調(diào)整指令的執(zhí)行順序，從而導(dǎo)致指令以不同的方式被執(zhí)行。通過使用適當(dāng)?shù)耐綑C(jī)制，比如互斥鎖，我們可以確保自己能夠讀取到共享資源的最新版本。

互斥鎖是如何實現(xiàn)的呢？

互斥鎖的實現(xiàn)通常需要操作系統(tǒng)和硬件的支持。如果系統(tǒng)只有一個處理器，那么我們可以通過在某個線程持有鎖時禁止中斷的方式來實現(xiàn)互斥鎖。這樣一來，其他線程就不會干擾當(dāng)前線程的執(zhí)行，從而避免沖突。不過，這種做法并不理想，因為編寫不當(dāng)?shù)拇a可能會導(dǎo)致整個系統(tǒng)被阻塞，讓其他進(jìn)程和線程都無法正常運行。此外，如果系統(tǒng)有多個處理器，這種做法也是行不通的，因為其他線程可以在其他 CPU 上并行執(zhí)行。

互斥量的實現(xiàn)需要硬件的支持，以便能夠進(jìn)行原子的測試和設(shè)置操作。通過這種操作，某個執(zhí)行單元可以檢查內(nèi)存中的值是否與預(yù)期相符；如果相符，它就會將內(nèi)存中的值設(shè)置為“已鎖定”狀態(tài)。硬件確保了這一測試和設(shè)置操作的原子性——也就是說，在操作完成之前，其他執(zhí)行單元無法訪問該內(nèi)存位置。早期的硬件實現(xiàn)方式是通過封鎖整個總線來確保這種原子性，這樣其他處理器就無法同時使用該內(nèi)存。如果另一個執(zhí)行單元嘗試進(jìn)行同樣的操作，卻發(fā)現(xiàn)內(nèi)存已經(jīng)被設(shè)置為“已鎖定”狀態(tài)，那么操作系統(tǒng)會阻止該線程的繼續(xù)執(zhí)行，直到內(nèi)存狀態(tài)再次變?yōu)?ldquo;可用”為止。

互斥鎖與順序處理

當(dāng)然，當(dāng)協(xié)程的數(shù)量超過兩個時，我們也可以使用互斥鎖來解決同步問題。

請注意：使用互斥鎖會限制程序的并發(fā)程度。在互斥鎖被鎖定和解鎖之間的代碼段，任何時候都只能由一個協(xié)程來執(zhí)行，這樣一來，這段代碼就變成了順序執(zhí)行的代碼。正如我們在第一章中所看到的，根據(jù)阿姆達(dá)爾定律，順序執(zhí)行與并行執(zhí)行的比例會限制代碼的性能可擴(kuò)展性。因此，我們必須盡量減少互斥鎖被占用的時間。

如果在 countLetters() 函數(shù)開始時鎖定互斥鎖，而在函數(shù)執(zhí)行完畢后釋放它，會怎么樣呢？從下面的代碼可以看出：我們在調(diào)用該函數(shù)后立即鎖定互斥鎖，而在輸出完成后的消息后釋放互斥鎖。

package main

import (
    "fmt"
    "io"
    "net/http"
    "strings"
    "sync"
    "time"
)

const AllLetters = "abcdefghijklmnopqrstuvwxyz"

func main() {
    mutex := sync.Mutex{}
    var frequency = make([]int, 26)
    for i := 1000; i <= 1030; i++ {
        url := fmt.Sprintf("https://rfc-editor.org/rfc/rfc%d.txt", i)
        go CountLetters(url, frequency, &mutex)
    }
    time.Sleep(60 * time.Second)
    mutex.Lock()
    for i, c := range AllLetters {
        fmt.Printf("%c-%d ", c, frequency[i])
    }
    mutex.Unlock()
}

// CountLetters
// Note: this program us locking the entire goroutine with mutex on purpose to demonstrate
// bad placement of the lock and unlock. We fix this in the next listing
func CountLetters(url string, frequency []int, mutex *sync.Mutex) {
    mutex.Lock()
    resp, _ := http.Get(url)
    defer resp.Body.Close()
    if resp.StatusCode != 200 {
        panic("Server returning error status code: " + resp.Status)
    }
    body, _ := io.ReadAll(resp.Body)
    for _, b := range body {
        c := strings.ToLower(string(b))
        cIndex := strings.Index(AllLetters, c)
        if cIndex >= 0 {
            frequency[cIndex] += 1
        }
    }
    fmt.Println("Completed:", url, time.Now().Format("15:04:05"))
    mutex.Unlock()
}

通過這種方式使用互斥鎖，我們將原本的并發(fā)程序變成了順序執(zhí)行的程序。這樣一來，我們每次只能下載并處理一個網(wǎng)頁，因為整個程序的執(zhí)行過程被無謂地阻塞了。如果繼續(xù)運行這個程序，其耗時將與非并發(fā)版本的程序相同，只不過各網(wǎng)頁的執(zhí)行順序會是隨機(jī)的而已。

提示：在決定何時以及如何使用互斥鎖時，我們應(yīng)先明確需要保護(hù)哪些資源，并確定各個臨界區(qū)的起始點和結(jié)束點。之后，我們需要想辦法盡量減少“Lock()”和“Unlock()”調(diào)用的次數(shù)。

package listing4_5

import (
    "fmt"
    "io"
    "net/http"
    "strings"
    "sync"
    "time"
)

const AllLetters = "abcdefghijklmnopqrstuvwxyz"

func CountLetters(url string, frequency []int, mutex *sync.Mutex) {
    resp, _ := http.Get(url)
    defer resp.Body.Close()
    if resp.StatusCode != 200 {
        panic("Server returning error code: " + resp.Status)
    }
    body, _ := io.ReadAll(resp.Body)
    mutex.Lock()
    for _, b := range body {
        c := strings.ToLower(string(b))
        cIndex := strings.Index(AllLetters, c)
        if cIndex >= 0 {
            frequency[cIndex] += 1
        }
    }
    mutex.Unlock()
    fmt.Println("Completed:", url, time.Now().Format("15:04:05"))
}

在這個代碼版本中，下載過程——也就是程序中耗時最長的部分——會以并行方式執(zhí)行。而快速處理字母計數(shù)的任務(wù)則會在順序模式下進(jìn)行。通過這種方式，我們能夠最大限度地提升程序的可擴(kuò)展性：因為我們只對那些執(zhí)行速度遠(yuǎn)遠(yuǎn)快于其他部分的代碼段使用鎖機(jī)制。

程序的執(zhí)行過程如圖所示。同樣，為了便于觀察，下載和處理所需的時間比例被夸大了。實際上，處理所需的時間遠(yuǎn)遠(yuǎn)少于下載網(wǎng)頁所需的時間，因此加速效果更為顯著。實際上，在我們的 main()函數(shù)中，可以將等待時間縮短到幾秒鐘左右（之前是 60 秒）。

第二種解決方案比我們的第一次嘗試要快得多。當(dāng)鎖定需要處理的代碼量較小時，我們完成任務(wù)的效率會更高。這里的經(jīng)驗法則是：盡量減少持有互斥鎖的時間，同時也要減少互斥鎖的調(diào)用次數(shù)?；叵胍幌掳⒛愤_(dá)爾定律就不難理解這一點：如果代碼中并行處理部分所占的時間更長，那么整體處理速度就會更快，系統(tǒng)的擴(kuò)展性也會更好。

非阻塞式互斥鎖

當(dāng)一個協(xié)程調(diào)用 Lock() 函數(shù)時，如果互斥鎖已被其他協(xié)程占用，那么該協(xié)程將會被阻塞。這就是所謂的“阻塞函數(shù)”：該協(xié)程的執(zhí)行會停止，直到有另一個協(xié)程調(diào)用 Unlock() 函數(shù)為止。在某些應(yīng)用中，我們會……

或許不必阻塞該協(xié)程，而是可以先執(zhí)行其他任務(wù)，然后再嘗試鎖定互斥量并訪問臨界區(qū)。

正因如此，Go 語言中的互斥鎖提供了另一個名為 TryLock()的函數(shù)。調(diào)用這個函數(shù)時，我們可能會遇到兩種結(jié)果之一：

• 鎖是可用的，這種情況下我們就能獲取到它，此時函數(shù)會返回布爾值“true”。
• 該鎖當(dāng)前無法使用，因為另一個協(xié)程正在使用這個互斥鎖。因此，該函數(shù)會立即返回，返回值為布爾值“false”。

非阻塞機(jī)制的用途

在 Go 語言的 1.18 版本中，新增了用于處理互斥鎖的 TryLock()函數(shù)。不過，關(guān)于這種非阻塞調(diào)用方式的實用示例其實很少見。這是因為在 Go 中，創(chuàng)建一個 goroutine 的成本遠(yuǎn)遠(yuǎn)低于在其他語言中創(chuàng)建內(nèi)核級線程的成本。如果互斥鎖不可用，讓 goroutine 做其他事情并沒有太大意義。因為在這種情況下，我們完全可以另開一個 goroutine 來執(zhí)行任務(wù)，而無需等待互斥鎖被釋放。實際上，Go 的互斥鎖文檔中也提到了這一點（來源：pkg.go.dev/sync#Mutex.TryLock）。

需要注意的是，雖然確實存在正確使用 TryLock 的情況，但這種情況相當(dāng)罕見。通常來說，使用 TryLock 往往意味著在多線程編程中存在更嚴(yán)重的問題。

使用 TryLock()的一個例子是：當(dāng)需要監(jiān)控某個任務(wù)的執(zhí)行進(jìn)度時，又不希望干擾該任務(wù)的正常進(jìn)行。如果使用普通的 Lock()機(jī)制，由于系統(tǒng)中有很多其他 goroutine 也在試圖獲取鎖，那么就會給互斥鎖帶來不必要的負(fù)擔(dān)，而這一切都是為了實現(xiàn)監(jiān)控功能而已。而使用 TryLock()的話，如果此時有別的 goroutine 正在占用互斥鎖，那么監(jiān)控用的 goroutine 就可以選擇稍后再試。這就好比去郵局處理一件不緊急的事務(wù)時，看到門口排著長隊，就決定改天再來辦理吧。

我們可以修改我們的字母頻率統(tǒng)計程序，讓主 goroutine 在下載數(shù)據(jù)和處理文檔的過程中，定期檢查頻率表的內(nèi)容。

package main

import (
  "fmt"
  "io"
  "net/http"
  "strings"
  "sync"
  "time"
)

func main() {
  mutex := sync.Mutex{}
  var frequency = make([]int, 26)
  for i := 2000; i <= 2200; i++ {
    url := fmt.Sprintf("https://rfc-editor.org/rfc/rfc%d.txt", i)
    go CountLetters(url, frequency, &mutex)
  }
  for i := 0; i < 100; i++ {
    time.Sleep(100 * time.Millisecond)
    if mutex.TryLock() {
      for i, c := range AllLetters {
        fmt.Printf("%c-%d ", c, frequency[i])
      }
      fmt.Println()
      mutex.Unlock()
    } else {
      fmt.Println("Mutex already being used")
    }
  }
}

const AllLetters = "abcdefghijklmnopqrstuvwxyz"

func CountLetters(url string, frequency []int, mutex *sync.Mutex) {
  resp, _ := http.Get(url)
  defer resp.Body.Close()
  if resp.StatusCode != 200 {
    panic("Server returning error code: " + resp.Status)
  }
  body, _ := io.ReadAll(resp.Body)
  mutex.Lock()
  for _, b := range body {
    c := strings.ToLower(string(b))
    cIndex := strings.Index(AllLetters, c)
    if cIndex >= 0 {
      frequency[cIndex] += 1
    }
  }
  mutex.Unlock()
  fmt.Println("Completed:", url, time.Now().Format("15:04:05"))
}

當(dāng)我們運行代碼時，可以從輸出結(jié)果中看到：main()協(xié)程試圖獲取鎖以打印出頻率表。有時它能成功獲取鎖并完成打??；而有時則無法獲取鎖，此時它需要等待 100 毫秒后再嘗試一次。

利用讀寫互斥鎖提升性能

有時候，互斥鎖的約束過于嚴(yán)格。可以把互斥鎖看作是一種“粗暴”的工具：它通過阻止多個線程同時執(zhí)行來解決問題。在任何時刻，都只能有一個 goroutine 能夠進(jìn)入受互斥鎖保護(hù)的代碼段。這種方式確實能有效避免競爭條件帶來的問題，但對于某些應(yīng)用來說，它可能會不必要地限制性能和可擴(kuò)展性。讀寫互斥鎖則是對標(biāo)準(zhǔn)互斥鎖的一種改進(jìn)：只有在我們需要更新共享資源時，它才會阻止其他線程的訪問。使用讀寫互斥鎖后，那些以讀取操作為主的程序的性能可以得到提升，因為在這種情況下，對共享數(shù)據(jù)的讀取操作遠(yuǎn)遠(yuǎn)多于更新操作。

Go 語言中的讀寫互斥鎖

如果所有處理客戶請求的協(xié)程都能以非獨占的方式訪問該數(shù)據(jù)結(jié)構(gòu)，那么它們就可以在需要時同時讀取數(shù)據(jù)。這樣可以提高性能，因為這樣多個僅負(fù)責(zé)讀取共享數(shù)據(jù)的協(xié)程就能同時訪問該數(shù)據(jù)。只有當(dāng)需要更新數(shù)據(jù)時，才會對共享數(shù)據(jù)的訪問進(jìn)行限制。在這個例子中，數(shù)據(jù)的更新頻率很低（每秒幾次），而讀取數(shù)據(jù)的頻率則非常高（每秒數(shù)千次）。因此，采用一種允許多個協(xié)程同時讀取數(shù)據(jù)、但只能有單個協(xié)程進(jìn)行寫入的機(jī)制，會更有利。

這就是讀寫鎖的作用。當(dāng)我們只需讀取共享資源而不對其進(jìn)行修改時，讀寫鎖允許多個協(xié)程同時執(zhí)行只讀操作。而當(dāng)我們需要修改共享資源時，負(fù)責(zé)寫入操作的協(xié)程會請求獲得寫鎖，從而實現(xiàn)對該資源的獨占訪問。這一原理在上圖中有所體現(xiàn)：圖的左側(cè)顯示，讀鎖允許多個協(xié)程同時進(jìn)行讀取操作，但會阻止任何寫入操作；而在圖的右側(cè)，一旦獲得了寫鎖，那么所有的讀取和寫入操作都會被阻止，這與普通互斥鎖的機(jī)制類似。

總結(jié)

• 互斥鎖可以用來保護(hù)代碼中的關(guān)鍵部分，防止它們被同時執(zhí)行。
• 我們可以通過在關(guān)鍵代碼段的開始處調(diào)用 Lock()函數(shù)，在結(jié)束處調(diào)用 unLock()函數(shù)來使用互斥鎖保護(hù)這些關(guān)鍵部分。
• 如果鎖定的時間過長，那么原本可以并行執(zhí)行的代碼就會變成順序執(zhí)行，從而降低性能。
• 我們可以通過調(diào)用 TryLock() 函數(shù)來檢測某個互斥鎖是否已被鎖定。
• 讀寫互斥鎖能夠提升那些以讀取操作為主的應(yīng)用程序的性能。
• 讀寫互斥鎖允許多個讀取者協(xié)程同時執(zhí)行臨界代碼段，同時確保只有單個寫入者協(xié)程能夠訪問該資源。
• 我們可以使用一個計數(shù)器以及兩個普通的互斥鎖來構(gòu)建一種“以讀取操作為主的讀寫互斥機(jī)制”。

到此這篇關(guān)于Go 語言使用互斥鎖進(jìn)行同步的文章就介紹到這了,更多相關(guān)Go 互斥鎖同步內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

最新評論