Go基礎系列：Go實現工做池的兩種方式

worker pool簡介

worker pool其實就是線程池thread pool。對於go來講，直接使用的是goroutine而非線程，不過這裏仍然以線程來解釋線程池。

在線程池模型中，有2個隊列一個池子：任務隊列、已完成任務隊列和線程池。其中已完成任務隊列可能存在也可能不存在，依據實際需求而定。

只要有任務進來，就會放進任務隊列中。只要線程執行完了一個任務，就將任務放進已完成任務隊列，有時候還會將任務的處理結果也放進已完成隊列中。

worker pool中包含了一堆的線程(worker，對go而言每一個worker就是一個goroutine)，這些線程嗷嗷待哺，等待着爲它們分配任務，或者本身去任務隊列中取任務。取得任務後更新任務隊列，而後執行任務，並將執行完成的任務放進已完成隊列。

下圖來自wiki：

在Go中有兩種方式能夠實現工做池：傳統的互斥鎖、channel。

傳統互斥鎖機制的工做池

假設Go中的任務的定義形式爲：

type Task struct {
    ...
}

每次有任務進來時，都將任務放在任務隊列中。

使用傳統的互斥鎖方式實現，任務隊列的定義結構大概以下：

type Queue struct{
    M     sync.Mutex
    Tasks []Task
}

而後在執行任務的函數中加上Lock()和Unlock()。例如：

func Worker(queue *Queue) {
    for {
        // Lock()和Unlock()之間的是critical section
        queue.M.Lock()
        // 取出任務
        task := queue.Tasks[0]
        // 更新任務隊列
        queue.Tasks = queue.Tasks[1:]
        queue.M.Unlock()
        // 在此goroutine中執行任務
        process(task)
    }
}

假如在線程池中激活了100個goroutine來執行Worker()。Lock()和Unlock()保證了在同一時間點只能有一個goroutine取得任務並隨之更新任務列表，取任務和更新任務隊列都是critical section中的代碼，它們是具備原子性。而後這個goroutine能夠執行本身取得的任務。於此同時，其它goroutine能夠爭奪互斥鎖，只要爭搶到互斥鎖，就能夠取得任務並更新任務列表。當某個goroutine執行完process(task)，它將由於for循環再次參與互斥鎖的爭搶。

上面只是給出了一點主要的代碼段，要實現完整的線程池，還有不少額外的代碼。

經過互斥鎖，上面的一切操做都是線程安全的。但問題在於加鎖/解鎖的機制比較重量級，當worker(即goroutine)的數量足夠多，鎖機制的實現將出現瓶頸。

經過buffered channel實現工做池

在Go中，也能用buffered channel實現工做池。

示例代碼很長，因此這裏先拆分解釋每一部分，最後給出完整的代碼段。

在下面的示例中，每一個worker的工做都是計算每一個數值的位數相加之和。例如給定一個數值234，worker則計算2+3+4=9。這裏交給worker的數值是隨機生成的[0,999)範圍內的數值。

這個示例有幾個核心功能須要先解釋，也是經過channel實現線程池的通常功能：

• 建立一個task buffered channel，並經過allocate()函數將生成的任務存放到task buffered channel中
  
• 建立一個goroutine pool，每一個goroutine監聽task buffered channel，並從中取出任務
  
• goroutine執行任務後，將結果寫入到result buffered channel中
  
• 從result buffered channel中取出計算結果並輸出

首先，建立Task和Result兩個結構，並建立它們的通道：

type Task struct {
    ID      int
    randnum int
}

type Result struct {
    task    Task
    result  int
}

var tasks = make(chan Task, 10)
var results = make(chan Result, 10)

這裏，每一個Task都有本身的ID，以及該任務將要被worker計算的隨機數。每一個Result都包含了worker的計算結果result以及這個結果對應的task，這樣從Result中就能夠取出任務信息以及計算結果。

另外，兩個通道都是buffered channel，容量都是10。每一個worker都會監聽tasks通道，並取出其中的任務進行計算，而後將計算結果和任務自身放進results通道中。

而後是計算位數之和的函數process()，它將做爲worker的工做任務之一。

func process(num int) int {
    sum := 0
    for num != 0 {
        digit := num % 10
        sum += digit
        num /= 10
    }
    time.Sleep(2 * time.Second)
    return sum
}

這個計算過程其實很簡單，但隨後還睡眠了2秒，用來僞裝執行一個計算任務是須要一點時間的。

而後是worker()，它監聽tasks通道並取出任務進行計算，並將結果放進results通道。

func worker(wg *WaitGroup){
    defer wg.Done()
    for task := range tasks {
        result := Result{task, process(task.randnum)}
        results <- result
    }
}

上面的代碼很容易理解，只要tasks channel不關閉，就會一直監聽該channel。須要注意的是，該函數使用指針類型的*WaitGroup做爲參數，不能直接使用值類型的WaitGroup做爲參數，這樣會使得每一個worker都有一個本身的WaitGroup。

而後是建立工做池的函數createWorkerPool()，它有一個數值參數，表示要建立多少個worker。

func createWorkerPool(numOfWorkers int) {
    var wg sync.WaitGroup
    for i := 0; i < numOfWorkers; i++ {
        wg.Add(1)
        go worker(&wg)
    }
    wg.Wait()
    close(results)
}

建立工做池時，首先建立一個WaitGroup的值wg，這個wg被工做池中的全部goroutine共享，每建立一個goroutine都wg.Add(1)。建立完全部的goroutine後等待全部的groutine都執行完它們的任務，只要有一個任務尚未執行完，這個函數就會被Wait()阻塞。當全部任務都執行完成後，關閉results通道，由於沒有結果再須要向該通道寫了。

固然，這裏是否須要關閉results通道，是由稍後的range迭代這個通道決定的，不關閉這個通道會一直阻塞range，最終致使死鎖。

工做池部分已經完成了。如今須要使用allocate()函數分配任務：生成一大堆的隨機數，而後將Task放進tasks通道。該函數有一個表明建立任務數量的數值參數：

func allocate(numOfTasks int) {
    for i := 0; i < numOfTasks; i++ {
        randnum := rand.Intn(999)
        task := Task{i, randnum}
        tasks <- task
    }
    close(tasks)
}

注意，最後須要關閉tasks通道，由於全部任務都分配完以後，沒有任務再須要分配。固然，這裏之因此須要關閉tasks通道，是由於worker()中使用了range迭代tasks通道，若是不關閉這個通道，worker將在取完全部任務後一直阻塞，最終致使死鎖。

再接着的是取出results通道中的結果進行輸出，函數名爲getResult()：

func getResult(done chan bool) {
    for result := range results {
        fmt.Printf("Task id %d, randnum %d , sum %d\n", result.task.id, result.task.randnum, result.result)
    }
    done <- true
}

getResult()中使用了一個done參數，這個參數是一個信號通道，用來表示results中的全部結果都取出來並處理完成了，這個通道不必定要用bool類型，任何類型皆可，它不用來傳數據，僅用來返回可讀，因此上面直接close(done)的效果也同樣。經過下面的main()函數，就能理解done信號通道的做用。

最後還差main()函數：

func main() {
    // 記錄起始終止時間，用來測試完成全部任務耗費時長
    startTime := time.Now()
    
    numOfWorkers := 20
    numOfTasks := 100
    // 建立任務到任務隊列中
    go allocate(numOfTasks)
    // 建立工做池
    go createWorkerPool(numOfWorkers)
    // 取得結果
    var done = make(chan bool)
    go getResult(done)

    // 若是results中還有數據，將阻塞在此
    // 直到發送了信號給done通道
    <- done
    endTime := time.Now()
    diff := endTime.Sub(startTime)
    fmt.Println("total time taken ", diff.Seconds(), "seconds")
}

上面分配了20個worker，這20個worker總共須要處理的任務數量爲100。但注意，不管是tasks仍是results通道，容量都是10，意味着任務隊列最長只能是10個任務。

下面是完整的代碼段：

package main

import (
    "fmt"
    "math/rand"
    "sync"
    "time"
)

type Task struct {
    id      int
    randnum int
}
type Result struct {
    task   Task
    result int
}

var tasks = make(chan Task, 10)
var results = make(chan Result, 10)

func process(num int) int {
    sum := 0
    for num != 0 {
        digit := num % 10
        sum += digit
        num /= 10
    }
    time.Sleep(2 * time.Second)
    return sum
}
func worker(wg *sync.WaitGroup) {
    defer wg.Done()
    for task := range tasks {
        result := Result{task, process(task.randnum)}
        results <- result
    }
}
func createWorkerPool(numOfWorkers int) {
    var wg sync.WaitGroup
    for i := 0; i < numOfWorkers; i++ {
        wg.Add(1)
        go worker(&wg)
    }
    wg.Wait()
    close(results)
}
func allocate(numOfTasks int) {
    for i := 0; i < numOfTasks; i++ {
        randnum := rand.Intn(999)
        task := Task{i, randnum}
        tasks <- task
    }
    close(tasks)
}
func getResult(done chan bool) {
    for result := range results {
        fmt.Printf("Task id %d, randnum %d , sum %d\n", result.task.id, result.task.randnum, result.result)
    }
    done <- true
}
func main() {
    startTime := time.Now()
    numOfWorkers := 20
    numOfTasks := 100

    var done = make(chan bool)
    go getResult(done)
    go allocate(numOfTasks)
    go createWorkerPool(numOfWorkers)
    // 必須在allocate()和getResult()以後建立工做池
    <-done
    endTime := time.Now()
    diff := endTime.Sub(startTime)
    fmt.Println("total time taken ", diff.Seconds(), "seconds")
}

執行結果：

Task id 19, randnum 914 , sum 14
Task id 9, randnum 150 , sum 6
Task id 15, randnum 215 , sum 8
............
Task id 97, randnum 315 , sum 9
Task id 99, randnum 641 , sum 11
total time taken  10.0174705 seconds

總共花費10秒。

能夠試着將任務數量、worker數量修改修改，看看它們的性能比例狀況。例如，將worker數量設置爲99，將須要4秒，將worker數量設置爲10，將須要20秒。