鎖就像漏斗，將并發(fā)處理的多個線程變成串行化的模式，我們可以構(gòu)建一個支持成千上萬并發(fā)的系統(tǒng)，但是如果鎖處理的不好會嚴(yán)重影響系統(tǒng)的性能，就像擁有多條車道的高速公路變成了單行道。

舉個例子，假如我們使用go的map來實(shí)現(xiàn)一個簡單的緩存，由于map不是并發(fā)安全,所以我們還要借助sync包的鎖來保證并發(fā)安全，于是我們很容易寫出下面這樣的代碼：

package simple_cache

import (
    "sync"
)

type Cache struct {
    items map[string][]byte
    lock  *sync.RWMutex
}

func New() *Cache {
    return &Cache{
        items: make(map[string][]byte, 2000),
        lock:  new(sync.RWMutex),
    }
}

func (c *Cache) Get(key string) []byte {
    // 取數(shù)據(jù)只要加讀鎖
    c.lock.RLock()
    defer c.lock.RUnlock()
    return c.items[key]
}

func (c *Cache) Set(key string, data []byte) {
    c.lock.Lock()
    defer c.lock.Unlock()
    c.items[key] = data
}

這段代碼考慮到了鎖其實(shí)已經(jīng)算是不錯了，但是每次調(diào)用set()方法去設(shè)置緩存值的時候不僅將并發(fā)讀寫變成了串行化的模式，就連get()方法也會被阻塞住。在實(shí)際生產(chǎn)中使用這段代碼作為緩存的時候，map中會緩存大量數(shù)據(jù)，set()調(diào)用可能會很頻繁，而且在set()內(nèi)還需要判斷緩存的容量是否足夠，這些都會使鎖的時間變長。

然后我們不得不考慮如何優(yōu)化一下鎖的性能。上面代碼的問題是每次set()都鎖住了整個map,于是我們就想到能不能只鎖住一部分，這樣就能降低鎖對性能的消耗。我們可以把原先這個大的緩存分成若干個小的分片，每個分片就是原先的一個Cache,然后再將這些分片放入一個大的map中，根據(jù)緩存key值通過hash計算后的值找到對應(yīng)的分片。對上面代碼改造如下：

package simple_cache

import (
    "crypto/sha1"
    "fmt"
    "sync"
)

type Cache map[string]*ShardCache

type ShardCache struct {
    items map[string][]byte
    lock  *sync.RWMutex
}

func NewCache() *Cache {
    cache := make(Cache, 256)
    for i := 0; i < 256; i++ {
        cache[fmt.Sprintf("%02x", i)] = &ShardCache{
            items: make(map[string][]byte, 2000),
            lock:  new(sync.RWMutex),
        }
    }
    return &cache
}

func (c Cache) getShard(key string) *ShardCache {
    hasher := sha1.New()
    hasher.Write([]byte(key))
        // 轉(zhuǎn)16進(jìn)制后取前兩位
    shardKey := fmt.Sprintf("%x", hasher.Sum(nil))[0:2]
    return c[shardKey]
}

func (c Cache) Get(key string) []byte {
    // 取數(shù)據(jù)只要加讀鎖
    shard := c.getShard(key)
    shard.lock.RLock()
    defer shard.lock.RUnlock()
    return shard.items[key]
}

func (c Cache) Set(key string, data []byte) {
    shard := c.getShard(key)
    shard.lock.Lock()
    shard.lock.Unlock()
    shard.items[key] = data
}

這里我們一共給緩存設(shè)置了256（16^2）個分片，對于任意的一個緩存key值經(jīng)過hash后通過fmt.Sprintf("%x", hasher.Sum(nil))[0:2]轉(zhuǎn)16進(jìn)制后取前兩位后都能在緩存中找到對應(yīng)的分片

其實(shí)像java里面的ConcurrencyHashmap已經(jīng)是這樣做的了，我們通過hash計算數(shù)據(jù)存儲的所在的分片，雖然消耗一點(diǎn)點(diǎn)計算資源但是解決了鎖粒度大導(dǎo)致的鎖性能問題，這是很值得的。

總結(jié)

• 通過對hash表分片，大鎖拆小鎖，降低鎖粒度，提高高并發(fā)情況下的鎖性能