map中因mutex使用不當導致的數(shù)據(jù)競爭

今天跟大家分享一個使用mutex在對slice或map的數(shù)據(jù)進行保護時容易被忽略的一個案例。

眾所周知，在并發(fā)程序中，對共享數(shù)據(jù)的訪問是經(jīng)常的事情，一般通過使用mutex對共享數(shù)據(jù)進行安全保護。當對slice和map使用mutex進行保護時有一個錯誤是經(jīng)常被忽略的。下面我們看一個具體的示例。

我們首先定義一個Cache結(jié)構(gòu)體，該結(jié)構(gòu)體用來緩存客戶的銀行卡的當前余額數(shù)據(jù)。該結(jié)構(gòu)體使用一個map來存儲，key是客戶的ID，value是客戶的余額。同時，有一個保護并發(fā)訪問的讀寫鎖變量。如下：

type Cache struct {    mu sync.RWMutex    balances map[string]float64}

接下來我們定義個AddBalance方法，該方法使用寫鎖來保護balances能被并發(fā)訪問。如下：

func (c *Cache) AddBalance(id string, balance float64) {    c.mu.Lock()    c.balances[id] = balance    c.mu.Unlock()}

同時，我們還實現(xiàn)了一個求所有客戶平均余額的函數(shù)。下面是其中的一種實現(xiàn)：

func (c *Cache) AverageBalance() float64 {    c.mu.RLock()    balances := c.balances    c.mu.RUnlock()        sum := 0.    for _, balance := range balances {        sum += balance    }        return sum / float64(len(balances))}

在該實現(xiàn)中，我們將c.balances拷貝到了一個本地變量中，然后就釋放了鎖。然后通過循環(huán)本地變量balances來計算所有客戶的總額。最后返回客戶的平均余額。以下是main中的代碼：

func main() {    cache := &Cache{        balances : make(map[string]float64),    }  
    go cache.AverageBalance()      go cache.AddBalance("ID-10", 100)}

那么，這種實現(xiàn)方式有什么問題嗎？如果我們使用-race運行，則會提示導致數(shù)據(jù)競爭。所以這里的問題處在哪里呢？

實際上，我們在之前講過map的底層數(shù)據(jù)結(jié)構(gòu)實際上是一些元信息加上一個指向buckets的數(shù)據(jù)指針。因此，當使用balances := c.balances時并沒有拷貝實際的數(shù)據(jù)。而只是拷貝了map的元信息而已。如下圖： 

這里只列出了map底層結(jié)構(gòu)體的關(guān)鍵字段，若想了解map底層的詳細結(jié)構(gòu)可以參考我之前的那篇 map的底層實現(xiàn)原理。由上圖可以看到兩個變量底層指向的數(shù)組實際上是同一個內(nèi)存地址。在并發(fā)中，兩個協(xié)程同時操作一個內(nèi)存地址的數(shù)據(jù)，而且其中一個是寫入操作，因此就造成了數(shù)據(jù)競爭。

那我們應(yīng)該如何避免該數(shù)據(jù)競爭呢？我們有兩種方式。

一種方式是當?shù)倪壿嬋绻臅r不是很大的話，可以擴大臨界區(qū)。如下：

func (c *Cache) AverageBalance() float64 {    c.mu.RLock()    defer c.mu.RUnlock()
    sum := 0    for _, balance := range c.balances {        sum += balance    }        return sum / float64(len(c.balances))}

在該實現(xiàn)中，整個函數(shù)都是臨界區(qū)，這樣也就避免了數(shù)據(jù)競爭。

第二種方式是將原來的map數(shù)據(jù)深度拷貝一份到本地變量。這種方式適用于迭代循環(huán)邏輯比較重（也就是耗時比較大）的場景。比如在迭代邏輯中會涉及到網(wǎng)絡(luò)IO（數(shù)據(jù)庫的讀寫等）。如下：

func (c *Cache) AverageBalance() float64 {    c.mu.RLock()    m := make(map[string]float64, len(c.balances))    for k, v := range c.balances {        m[k] = v    }    c.mu.RUnlock()
    sum := 0    for _, balance := range balances {        sum += balance    }        return sum / float64(len(c.balances))}

在這種實現(xiàn)方案中，一旦我們完成了深度拷貝，就將鎖給釋放。同時，迭代的邏輯在臨界區(qū)外實現(xiàn)。

總之，當我們使用互斥鎖時一定要格外注意臨界區(qū)。今天的分享就到這里了。

推薦閱讀

• 學到了！將緩沖 channel 當做 Mutex 來使用