高併發下map和chan實現的連結池的執行緒安全及效率

阿新 • • 發佈：2018-12-13

1.背景

~~上一次blog寫著寫著崩掉了，這次一定寫完一節儲存一節。~~ 目前從事go語言的後臺開發，在叢集通訊時需要用到thrift的rpc。由於叢集間通訊非常頻繁且併發需求很高，所以只能採用連線池的形式。由於叢集規模是有限的，每個節點都需要儲存平行節點的連線，所以連結池的實現方式應該是map[host]chan conn。在go語言中，我們知道channel是執行緒安全的，但map卻不是執行緒安全的。所以我們需要適當的加鎖來保證其執行緒安全同時兼顧效率。

2. 連結池的一般設計

新建連結時，需要給定目標地址
獲取連結時，需要給定目標地址，若連結池存在連結，則從連結池返回，若不存在連結，則新建連結返回

連結池中存放的連結總是空閒連結
連線使用完後需放回連結池
放回連結池需要給定目標地址
放回連結池時若連結池已滿，則關閉該連結並將其交給gc

3.連結池的一般定義

type factory func(host string) conn
type conn interface {
   Close() error
}
type pool struct {
   m map[string]chan conn
   mu sync.RWMutex
   fact factory
}

以上是一個通用連結池的實現，用了channel和讀寫鎖，暫時不考慮連結超時等問題。我們的目的是探索這個連結池在高併發情況下的執行緒安全和get，put效率問題。所以下來我們給出實驗主程式。

4.測試主程式

測試主程式如下所示 pt是記錄的get，put次數，採用原子操作進行累加，其耗時忽略不計。 hosts為叢集規模，也是map的大小，一般來說不會太大。 threadnum為併發的協程數為了方便，此處直接使用net.dial作為工廠方法實現每一個協程是一個死迴圈，不斷地進行get，put操作。每次操作會使pts加1。

func main(){
   var pts uint64
   p := &pool{
      m :make(map[string]chan conn),
      mu:sync.RWMutex{},
      fact:func(target string)conn{
         c,_ :=net.Dial("","8080")
         return c
      },
   }
   //列印執行緒，列印get，put效率
   be := time.Now()
   go func (){
      for true{
         //此處先休眠一秒是為了避免第一次時差計算為0導致的除法錯誤
         time.Sleep(1 *time.Second)
         cost := time.Since(be) / time.Second
         println(atomic.LoadUint64(&pts)/uint64(cost),"pt/s")
      }
   }()
   time.Sleep(1*time.Second)
   //列印執行緒完，此處等待一秒是為對應列印執行緒第一次休眠，儘量減少誤差


   //叢集規模
   hosts := []string{"192.168.0.1","192.168.0.2","192.168.0.3","192.168.0.4"}
   //併發執行緒數量
   threadnum := 1
   for i:=0;i<threadnum;i++{
      go func(){
         for true{
            target := hosts[rand.Int() % len(hosts)]
            conn := p.Get(target)
            //------------------使用連線開始
            //time.Sleep(1*time.Nanosecond)
            //------------------使用連線完畢
            p.Put(target,conn)
            atomic.AddUint64(&pts,1)
         }
      }()
   }
   time.Sleep(100 * time.Second)
}

5.單協程情況下的效率

5.1 單協程get & put實現

單協程模式下，我們不必考慮執行緒安全的問題，也就不必加鎖。此時的get，put實現如下

func (p *pool)Get(host string) (c conn){
   if _,ok := p.m[host];!ok{
      p.m[host] = make(chan conn,100)
   }
   select {
   case c  = <- p.m[host]:
      {}
   default:
      c = p.New(host)
   }
   return
}
func (p *pool)Put(host string,c conn){
   select {
   case p.m[host] <- c:
      {}
   default:
      c.Close()
   }
}
func (p *pool)New(host string)conn{
   return p.fact(host)
}

5.2 測試結果

我們設定threadnum為1，測試結果如下。其速度大概在5,000,000 次/秒在這裡插入圖片描述

6.併發情況下效率-全寫鎖

6.1 全寫鎖的get & put 實現

為了保證併發情況下的執行緒安全，我們需要使用讀寫鎖，那麼對get和put操作究竟該如何加鎖呢，最安全的形式當然是全寫鎖的形式，單其效率肯定是最低的，因為這樣同一時刻總是隻有一個協程在進行寫或者讀。

func (p *pool)Get1(host string) (c conn){
   p.mu.Lock()
   defer p.mu.RLock()
   if _,ok := p.m[host];!ok{
      p.m[host] = make(chan conn,100)
   }
   select {
   case c  = <- p.m[host]:
      {}
   default:
      c = p.New(host)
   }
   return
}
func (p *pool)Put1(host string,c conn){
   p.mu.Lock()
   defer p.mu.Unlock()
   select {
   case p.m[host] <- c:
      {}
   default:
      c.Close()
   }
}

6.2 測試結果

#6.2.1 全寫鎖下的多協程測試結果

我們設定threadnum為4，測試結果如下，其速度大概在1,000,000次/秒在這裡插入圖片描述

6.2.2 全寫鎖下單協程測試結果

如果我們將threadnum設定為1，再次測試，其速度為2,800,00次/秒。可以看到，多協程會降低效率，因為協程間切換也會有時間消耗。但我們經常聽說多協程會提高執行速度，這也是對的，那麼什麼時候多協程會提高運速度呢，這就是我說的連結使用時間的問題，當連線使用時間大於鎖競爭和協程切換時間的時候，我們用多協程會提高效率。而實際使用中，連線的使用時間總是存在的且一般都大於鎖競爭時間和協程切換時間。在這裡插入圖片描述

6.2.3 單協程下存在連結使用時間的的測試結果

在主程式中，我們在get和put間加上休眠時間，此處設定休眠時間為1毫秒即連結使用1毫秒後放連結池。同時協程數設定為1。單協程情況下，其速度大概如下500次/秒。可以看到實際的效率大幅度降低。在這裡插入圖片描述

6.2.4 多協程下存在連結使用時間的測試結果

同樣保持連結使用時間為1毫秒，協程數量設定為4，測試結果如下。其速度大概為2,000次/秒，剛好是單協程的4倍。所以實際情況下多協程的使用需要慎重考慮，並不是多協程一定能提高程式的處理速度，相反在某些情況下會降低程式的執行速度。由於本次測試的是連結池的效能和安全，接下來的測試不再新增連結使用時間，只單純的測試讀寫鎖和效率的問題。本小節算是一個附加測試。

在這裡插入圖片描述

7.併發情況下效率-讀寫鎖1

由於全寫鎖沒有實際的使用意義，所以我們需要使用讀寫鎖來提高效率，那麼如何保證執行緒安全新增讀寫鎖呢。首先對於我們的map結構來說，當有寫操作的時候，我們的讀操作應該是不可靠的，所以不能進行，當讀操作時，我們不希望有寫操作但其他協程也能同時讀取，這橋恰符合讀寫鎖的作用原理。當加寫鎖時，所有的讀寫均不可用當加讀鎖時，所有的寫操作不可用，讀操作可用

7.1 讀寫鎖1 get & put實現

考察我們的put程式，只有對map的讀，所以只需要加讀鎖，而在get中，包含了兩部分，第一次寫操作和第二次的讀操作，所以我們很簡單的我們想到，需要使用兩次鎖，第一次寫鎖，第二次讀鎖。

func (p *pool)Get2(host string) (c conn){
   p.mu.Lock()
   if _,ok := p.m[host];!ok{
      p.m[host] = make(chan conn,100)
   }
   p.mu.Unlock()
   
   p.mu.RLock()
   defer p.mu.RUnlock()
   select {
   case c  = <- p.m[host]:
      {}
   default:
      c = p.New(host)
   }
   return
}
func (p *pool)Put1(host string,c conn){
   p.mu.RLock()
   defer p.mu.RUnlock()
   select {
   case p.m[host] <- c:
      {}
   default:
      c.Close()
   }
}

7.2 測試結果

我們本來期望的是效率應該比全寫鎖要高一些，但實際情況是低一些，只有800,000次/秒。那問題出在哪裡呢。從程式上來看，get多了一次加鎖，所以導致鎖競爭次數比全寫鎖要高一些，但我們並不能減少鎖次數直接使用讀鎖，這樣是不安全的，程式也會報錯。所以我們給出另一種安全的讀寫鎖形式。在這裡插入圖片描述

8.併發情況下效率-讀寫鎖

8.1 讀寫鎖2 get & put實我們從實際的使用來看一下get程式，由於我們給定了hosts，所以其實對map的寫入操作只會進行四次，但後來每次進行get時都會加一次寫鎖，這是沒有必要的。仔細看一下第一次寫鎖，我們加的有些草率，因為首先會讀取一次map來判斷是否應該進行寫入操作，所以我們可以通過增加一次讀鎖，來減少後來的加寫鎖。當然有人會說為什麼不直接初始化map，這樣就沒有寫操作，這我也考慮過，但是叢集規模有可能會擴張並且會動態變化，直接初始化map會顯得有些刻意，並且通用性也不強，與其他模組會產生耦合。所以這種做法並沒有多少設計上的美感，相反會顯得比較low。我們給出第二種讀寫鎖如下。

func (p *pool)Get3(host string) (c conn){
   p.mu.RLock()
   if _,ok := p.m[host];!ok{
      p.mu.RUnlock()
      p.mu.Lock()
      p.m[host] = make(chan conn,100)
      p.mu.Unlock()
   }else{
      p.mu.RUnlock()
   }
   p.mu.RLock()
   defer p.mu.RUnlock()
   select {
   case c = <- p.m[host]:
      {}
   default:
      c = p.New(host)
   }
   return
}
func Put3(host string,c conn){
   p.mu.RLock()
   defer p.mu.RUnlock()
   select {
   case p.m[host] <- c:
      {}
   default:
      c.Close()
   }
}

8.2 測試結測試結果如下，其速度大概在3,400,000次/秒。是全寫鎖效能的4倍左右。