Golang中怎么實(shí)現(xiàn)一個(gè)工作池

本篇文章為大家展示了Golang中怎么實(shí)現(xiàn)一個(gè)工作池，內(nèi)容簡(jiǎn)明扼要并且容易理解，絕對(duì)能使你眼前一亮，通過(guò)這篇文章的詳細(xì)介紹希望你能有所收獲。

worker pool簡(jiǎn)介

worker pool其實(shí)就是線程池thread pool。對(duì)于go來(lái)說(shuō)，直接使用的是goroutine而非線程，不過(guò)這里仍然以線程來(lái)解釋線程池。

在線程池模型中，有2個(gè)隊(duì)列一個(gè)池子：任務(wù)隊(duì)列、已完成任務(wù)隊(duì)列和線程池。其中已完成任務(wù)隊(duì)列可能存在也可能不存在，依據(jù)實(shí)際需求而定。

只要有任務(wù)進(jìn)來(lái)，就會(huì)放進(jìn)任務(wù)隊(duì)列中。只要線程執(zhí)行完了一個(gè)任務(wù)，就將任務(wù)放進(jìn)已完成任務(wù)隊(duì)列，有時(shí)候還會(huì)將任務(wù)的處理結(jié)果也放進(jìn)已完成隊(duì)列中。

worker pool中包含了一堆的線程(worker，對(duì)go而言每個(gè)worker就是一個(gè)goroutine)，這些線程嗷嗷待哺，等待著為它們分配任務(wù)，或者自己去任務(wù)隊(duì)列中取任務(wù)。取得任務(wù)后更新任務(wù)隊(duì)列，然后執(zhí)行任務(wù)，并將執(zhí)行完成的任務(wù)放進(jìn)已完成隊(duì)列。

下圖來(lái)自wiki：

在Go中有兩種方式可以實(shí)現(xiàn)工作池：傳統(tǒng)的互斥鎖、channel。

傳統(tǒng)互斥鎖機(jī)制的工作池

假設(shè)Go中的任務(wù)的定義形式為：

type Task struct {    ...}

每次有任務(wù)進(jìn)來(lái)時(shí)，都將任務(wù)放在任務(wù)隊(duì)列中。

使用傳統(tǒng)的互斥鎖方式實(shí)現(xiàn)，任務(wù)隊(duì)列的定義結(jié)構(gòu)大概如下：

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

然后在執(zhí)行任務(wù)的函數(shù)中加上Lock()和Unlock()。例如：

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

假如在線程池中激活了100個(gè)goroutine來(lái)執(zhí)行Worker()。Lock()和Unlock()保證了在同一時(shí)間點(diǎn)只能有一個(gè)goroutine取得任務(wù)并隨之更新任務(wù)列表，取任務(wù)和更新任務(wù)隊(duì)列都是critical section中的代碼，它們是具有原子性。然后這個(gè)goroutine可以執(zhí)行自己取得的任務(wù)。于此同時(shí)，其它goroutine可以爭(zhēng)奪互斥鎖，只要爭(zhēng)搶到互斥鎖，就可以取得任務(wù)并更新任務(wù)列表。當(dāng)某個(gè)goroutine執(zhí)行完process(task)，它將因?yàn)閒or循環(huán)再次參與互斥鎖的爭(zhēng)搶。

上面只是給出了一點(diǎn)主要的代碼段，要實(shí)現(xiàn)完整的線程池，還有很多額外的代碼。

通過(guò)互斥鎖，上面的一切操作都是線程安全的。但問(wèn)題在于加鎖/解鎖的機(jī)制比較重量級(jí)，當(dāng)worker(即goroutine)的數(shù)量足夠多，鎖機(jī)制的實(shí)現(xiàn)將出現(xiàn)瓶頸。

通過(guò)buffered channel實(shí)現(xiàn)工作池

在Go中，也能用buffered channel實(shí)現(xiàn)工作池。

示例代碼很長(zhǎng)，所以這里先拆分解釋每一部分，最后給出完整的代碼段。

在下面的示例中，每個(gè)worker的工作都是計(jì)算每個(gè)數(shù)值的位數(shù)相加之和。例如給定一個(gè)數(shù)值234，worker則計(jì)算2+3+4=9。這里交給worker的數(shù)值是隨機(jī)生成的[0,999)范圍內(nèi)的數(shù)值。

這個(gè)示例有幾個(gè)核心功能需要先解釋，也是通過(guò)channel實(shí)現(xiàn)線程池的一般功能：

創(chuàng)建一個(gè)task buffered channel，并通過(guò)allocate()函數(shù)將生成的任務(wù)存放到task buffered channel中創(chuàng)建一個(gè)goroutine pool，每個(gè)goroutine監(jiān)聽(tīng)task buffered channel，并從中取出任務(wù)goroutine執(zhí)行任務(wù)后，將結(jié)果寫(xiě)入到result buffered channel中從result buffered channel中取出計(jì)算結(jié)果并輸出

首先，創(chuàng)建Task和Result兩個(gè)結(jié)構(gòu)，并創(chuàng)建它們的通道：

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)

這里，每個(gè)Task都有自己的ID，以及該任務(wù)將要被worker計(jì)算的隨機(jī)數(shù)。每個(gè)Result都包含了worker的計(jì)算結(jié)果result以及這個(gè)結(jié)果對(duì)應(yīng)的task，這樣從Result中就可以取出任務(wù)信息以及計(jì)算結(jié)果。

另外，兩個(gè)通道都是buffered channel，容量都是10。每個(gè)worker都會(huì)監(jiān)聽(tīng)tasks通道，并取出其中的任務(wù)進(jìn)行計(jì)算，然后將計(jì)算結(jié)果和任務(wù)自身放進(jìn)results通道中。

然后是計(jì)算位數(shù)之和的函數(shù)process()，它將作為worker的工作任務(wù)之一。

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

這個(gè)計(jì)算過(guò)程其實(shí)很簡(jiǎn)單，但隨后還睡眠了2秒，用來(lái)假裝執(zhí)行一個(gè)計(jì)算任務(wù)是需要一點(diǎn)時(shí)間的。

然后是worker()，它監(jiān)聽(tīng)tasks通道并取出任務(wù)進(jìn)行計(jì)算，并將結(jié)果放進(jìn)results通道。

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

上面的代碼很容易理解，只要tasks channel不關(guān)閉，就會(huì)一直監(jiān)聽(tīng)該channel。需要注意的是，該函數(shù)使用指針類型的*WaitGroup作為參數(shù)，不能直接使用值類型的WaitGroup作為參數(shù)，這樣會(huì)使得每個(gè)worker都有一個(gè)自己的WaitGroup。

然后是創(chuàng)建工作池的函數(shù)createWorkerPool()，它有一個(gè)數(shù)值參數(shù)，表示要?jiǎng)?chuàng)建多少個(gè)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)}

創(chuàng)建工作池時(shí)，首先創(chuàng)建一個(gè)WaitGroup的值wg，這個(gè)wg被工作池中的所有g(shù)oroutine共享，每創(chuàng)建一個(gè)goroutine都wg.Add(1)。創(chuàng)建完所有的goroutine后等待所有的groutine都執(zhí)行完它們的任務(wù)，只要有一個(gè)任務(wù)還沒(méi)有執(zhí)行完，這個(gè)函數(shù)就會(huì)被Wait()阻塞。當(dāng)所有任務(wù)都執(zhí)行完成后，關(guān)閉results通道，因?yàn)闆](méi)有結(jié)果再需要向該通道寫(xiě)了。

當(dāng)然，這里是否需要關(guān)閉results通道，是由稍后的range迭代這個(gè)通道決定的，不關(guān)閉這個(gè)通道會(huì)一直阻塞range，最終導(dǎo)致死鎖。

工作池部分已經(jīng)完成了?，F(xiàn)在需要使用allocate()函數(shù)分配任務(wù)：生成一大堆的隨機(jī)數(shù)，然后將Task放進(jìn)tasks通道。該函數(shù)有一個(gè)代表創(chuàng)建任務(wù)數(shù)量的數(shù)值參數(shù)：

func allocate(numOfTasks int) { for i := 0; i < numOfTasks; i++ {

randnum := rand.Intn(999)        task := Task{i, randnum}        tasks <- task    } close(tasks)}

注意，最后需要關(guān)閉tasks通道，因?yàn)樗腥蝿?wù)都分配完之后，沒(méi)有任務(wù)再需要分配。當(dāng)然，這里之所以需要關(guān)閉tasks通道，是因?yàn)閣orker()中使用了range迭代tasks通道，如果不關(guān)閉這個(gè)通道，worker將在取完所有任務(wù)后一直阻塞，最終導(dǎo)致死鎖。

再接著的是取出results通道中的結(jié)果進(jìn)行輸出，函數(shù)名為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()中使用了一個(gè)done參數(shù)，這個(gè)參數(shù)是一個(gè)信號(hào)通道，用來(lái)表示results中的所有結(jié)果都取出來(lái)并處理完成了，這個(gè)通道不一定要用bool類型，任何類型皆可，它不用來(lái)傳數(shù)據(jù)，僅用來(lái)返回可讀，所以上面直接close(done)的效果也一樣。通過(guò)下面的main()函數(shù)，就能理解done信號(hào)通道的作用。

最后還差main()函數(shù)：

func main() { // 記錄起始終止時(shí)間，用來(lái)測(cè)試完成所有任務(wù)耗費(fèi)時(shí)長(zhǎng) startTime := time.Now()        numOfWorkers := 20 numOfTasks := 100 // 創(chuàng)建任務(wù)到任務(wù)隊(duì)列中 go allocate(numOfTasks) // 創(chuàng)建工作池 go createWorkerPool(numOfWorkers) // 取得結(jié)果 var done = make(chan bool) go getResult(done) // 如果results中還有數(shù)據(jù)，將阻塞在此 // 直到發(fā)送了信號(hào)給done通道 <- done    endTime := time.Now()    diff := endTime.Sub(startTime)    fmt.Println("total time taken ", diff.Seconds(), "seconds")}

上面分配了20個(gè)worker，這20個(gè)worker總共需要處理的任務(wù)數(shù)量為100。但注意，無(wú)論是tasks還是results通道，容量都是10，意味著任務(wù)隊(duì)列最長(zhǎng)只能是10個(gè)任務(wù)。

下面是完整的代碼段：

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()之后創(chuàng)建工作池 <-done    endTime := time.Now()    diff := endTime.Sub(startTime)    fmt.Println("total time taken ", diff.Seconds(), "seconds")}

執(zhí)行結(jié)果：

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

總共花費(fèi)10秒。

可以試著將任務(wù)數(shù)量、worker數(shù)量修改修改，看看它們的性能比例情況。例如，將worker數(shù)量設(shè)置為99，將需要4秒，將worker數(shù)量設(shè)置為10，將需要20秒。

上述內(nèi)容就是Golang中怎么實(shí)現(xiàn)一個(gè)工作池，你們學(xué)到知識(shí)或技能了嗎？如果還想學(xué)到更多技能或者豐富自己的知識(shí)儲(chǔ)備，歡迎關(guān)注億速云行業(yè)資訊頻道。