在執行 redis-benchmark 時，如果不指定 -r 參數的話，它會請求相同的 key，在我們的設計中它會被互斥鎖擋住而降低吞吐量。

註：按照目前的設計，就算我們指定了 -r 參數，所有請求也都會被互斥鎖擋下來，這是因為 Go 的 Map 是非並行安全的，我們必須使用鎖保證同時只有一個 goroutine 來存取。

當我們使用 redis-benchmark -p 16379 -t SET -P 16 會得到如下的火焰圖：

互斥鎖耗費了約 40% 的 CPU 時間，而即便我們加入 -r 參數也會是差不多的結果。

註：redis-benchmark 加入 -r 參數的話，會使用隨機的 key，其格式為 key:{12 bytes number}，而 12 bytes number 會落在 $[0, keyspacelen]$ 之間。

我們可以從 http://localhost:6060/debug/pprof/block 中取得更多有關阻塞的資訊：

可以發現它消耗了 1464.62 秒在等待互斥鎖上。

註：雖然我們只測試物理時間的 30 秒，但此處的秒數是根據所有 goroutine 所等待的時間加總，因此會大於 30 秒。

條帶化鎖（Lock Striping）

條帶化鎖是一個並行程式設計中的模式：只鎖定一部份的資源，從而避免鎖住整個資源。

我們可以定義一些「槽位（Slot）」，並且用 hash 的方式決定 key 會被放到哪些槽位。如此一來，存取資源時只需要鎖定相應的槽位即可。

重構 repo.(*mapStorage)

// file: ./internal/repo/storage.go
func NewStorage() Storage {
	s := mapStorage{}
	for i := range lockStripeCount {
		s.storage[i] = make(map[string]object.Object)
	}

	return &s
}

const lockStripeCount = 16

type mapStorage struct {
	storage [lockStripeCount]map[string]object.Object
	stripes [lockStripeCount]sync.Mutex
}

我們定義了 const lockStripeCount = 16，這是一個在性能上相當平衡的選擇。

// file: ./internal/repo/storage.go
func (s *mapStorage) stripe(k string) uint64 { 
	return xxh3.HashString(k) & (lockStripeCount - 1) 
} 

使用 xxh3.HashString(k) 來對 key 做雜湊，它會回傳一個 uint64，而我們僅有 16 個槽位，因此可以用 hashed % 16 來分配要去哪個槽位。

註：這裡我用了一點 bitwise 的小技巧，hashed % 16 等價於 hashed & 15，因為 lockStripeCount 是 2 的冪。

重構 repo.(*mapStorage).Set() 與 repo.(*mapStorage).Get()

在 Get 與 Set 的過程中，我們需要選擇對應的槽位：

// file: internal/repo/storage.go
func (s *mapStorage) setString(param SetStringParam) error { 
	obj := param.Obj() 
  
	slot := s.stripe(obj.Key()) 
	s.stripes[slot].Lock() 
	defer s.stripes[slot].Unlock() 
  
	cur, exists := s.storage[slot][obj.Key()] 
// ...
}

func (s *mapStorage) Get(_ context.Context, k string) (v object.Object, err error) { 
	slot := s.stripe(k) 
	s.stripes[slot].Lock() 
	defer s.stripes[slot].Unlock() 
  
	var ok bool 
	if v, ok = s.storage[slot][k]; !ok { 
		return nil, fmt.Errorf("%w: miss", ErrNotFound) 
	} 
	if v.ExpiresAt() != nil && time.Now().After(*v.ExpiresAt()) { 
 		delete(s.storage[slot], k) // remove key from storage when expired 
		return nil, fmt.Errorf("%w: expired", ErrNotFound) 
	} 
  
	return 
 } 

結果

在重構後，我們可以取得如下的火焰圖：

CPU 時間從 47.33 秒（37.7%）降到 8.58 秒（4.7%）

並且阻塞時間從 1464.62 秒降到 22.53 秒。