并發(fā)編程包之 Errgroup

作者：Seekload 2021-08-05 07:58:22

Task 是需要處理單個(gè)工作單元；Worker 是一個(gè)簡(jiǎn)單的 worker 函數(shù)，用于執(zhí)行任務(wù)；而 Pool 用于創(chuàng)建、管理 workers。

本文轉(zhuǎn)載自微信公眾號(hào)「Golang來(lái)啦」，作者Seekload 。轉(zhuǎn)載本文請(qǐng)聯(lián)系Golang來(lái)啦公眾號(hào)。

四哥水平有限，如果有翻譯或理解錯(cuò)誤的點(diǎn)，煩請(qǐng)幫忙指出，感謝!

這是系列文章的第二篇，第一篇文章點(diǎn)擊這里查看。

原文如下：

基于 goroutine 和 channel 的并發(fā)特性，使得 Go 成為了強(qiáng)大的并發(fā)語(yǔ)言。上一篇文章，我們討論了如何構(gòu)建 workerPool 來(lái)提高程序的并發(fā)性能，換句話說(shuō)，避免耗盡系統(tǒng)資源。但那只是一個(gè)簡(jiǎn)單的示例，演示我們應(yīng)該如何實(shí)現(xiàn)。

基于對(duì)上一篇文章的學(xué)習(xí)，在這篇文章里面，我們將構(gòu)建一個(gè)健壯的解決方案，以便在任何其他應(yīng)用程序里面可以使用該方案。網(wǎng)絡(luò)上有其他復(fù)雜架構(gòu)的解決方案，比如使用調(diào)度器等等。實(shí)際上，我們并不需要這些復(fù)雜的設(shè)計(jì)，僅僅使用一個(gè)共享 channel 就可以解決問(wèn)題。我們一起來(lái)看下，該如何構(gòu)建呢?

代碼結(jié)構(gòu)

我們創(chuàng)建了一個(gè)通用的 workerPool 包，根據(jù)業(yè)務(wù)所需的并發(fā)性使用 worker 來(lái)處理任務(wù)。一起來(lái)看下目錄結(jié)構(gòu)：

workerpool 
├── pool.go 
├── task.go 
└── worker.go

workerpool 目錄在項(xiàng)目的根目錄下。Task 是需要處理單個(gè)工作單元;Worker 是一個(gè)簡(jiǎn)單的 worker 函數(shù)，用于執(zhí)行任務(wù);而 Pool 用于創(chuàng)建、管理 workers。

實(shí)現(xiàn)

先看下 Task 代碼：

// workerpool/task.go 
 
package workerpool 
 
import ( 
 "fmt" 
) 
 
type Task struct { 
 Err  error 
 Data interface{} 
 f    func(interface{}) error 
} 
 
func NewTask(f func(interface{}) error, data interface{}) *Task { 
 return &Task{f: f, Data: data} 
} 
 
func process(workerID int, task *Task) { 
 fmt.Printf("Worker %d processes task %v\n", workerID, task.Data) 
 task.Err = task.f(task.Data) 
}

Task 是一個(gè)簡(jiǎn)單的結(jié)構(gòu)體，保存處理任務(wù)所需要的一切數(shù)據(jù)。創(chuàng)建 task 時(shí)，傳遞了 Data 和待執(zhí)行函數(shù) f，process() 函數(shù)會(huì)處理任務(wù)。處理任務(wù)時(shí)，將 Data 作為參數(shù)傳遞給函數(shù) f，并將執(zhí)行結(jié)果保存在 Task.Err 里。

我們來(lái)看下 Worker 是如何處理任務(wù)的：

// workerpool/worker.go 
 
package workerpool 
 
import ( 
 "fmt" 
 "sync" 
) 
 
// Worker handles all the work 
type Worker struct { 
 ID       int 
 taskChan chan *Task 
} 
 
// NewWorker returns new instance of worker 
func NewWorker(channel chan *Task, ID int) *Worker { 
 return &Worker{ 
  ID:       ID, 
  taskChan: channel, 
 } 
} 
 
// Start starts the worker 
func (wr *Worker) Start(wg *sync.WaitGroup) { 
 fmt.Printf("Starting worker %d\n", wr.ID) 
 
 wg.Add(1) 
 go func() { 
  defer wg.Done() 
  for task := range wr.taskChan { 
   process(wr.ID, task) 
  } 
 }() 
}

我們創(chuàng)建了一個(gè)小巧的 Worker 結(jié)構(gòu)體，包含 worker ID 和一個(gè)保存待處理任務(wù)的 channel。在 Start() 方法里，使用 for range 從 taskChan 讀取任務(wù)并處理。可以想象的到，多個(gè) worker 可以并發(fā)地執(zhí)行任務(wù)。

workerPool

我們通過(guò)實(shí)現(xiàn) Task 和 Worker 來(lái)處理任務(wù)，但是好像還缺點(diǎn)什么東西，誰(shuí)負(fù)責(zé)生成這些 worker 并將任務(wù)發(fā)送給它們?答案是：Worker Pool。

// workerpoo/pool.go 
 
package workerpool 
 
import ( 
 "fmt" 
 "sync" 
 "time" 
) 
 
// Pool is the worker pool 
type Pool struct { 
 Tasks   []*Task 
 
 concurrency   int 
 collector     chan *Task 
 wg            sync.WaitGroup 
} 
 
// NewPool initializes a new pool with the given tasks and 
// at the given concurrency. 
func NewPool(tasks []*Task, concurrency int) *Pool { 
 return &Pool{ 
  Tasks:       tasks, 
  concurrency: concurrency, 
  collector:   make(chan *Task, 1000), 
 } 
} 
 
// Run runs all work within the pool and blocks until it's 
// finished. 
func (p *Pool) Run() { 
 for i := 1; i <= p.concurrency; i++ { 
  worker := NewWorker(p.collector, i) 
  worker.Start(&p.wg) 
 } 
 
 for i := range p.Tasks { 
  p.collector <- p.Tasks[i] 
 } 
 close(p.collector) 
 
 p.wg.Wait() 
}

上面的代碼，pool 保存了所有待處理的任務(wù)，并且生成與 concurrency 數(shù)量一致的 goroutine，用于并發(fā)地處理任務(wù)。workers 之間共享緩存 channel -- collector。

所以，當(dāng)我們把這個(gè)工作池跑起來(lái)時(shí)，可以生成滿足所需數(shù)量的 worker，workers 之間共享 collector channel。接著，使用 for range 讀取 tasks，并將讀取到的 task 寫(xiě)入 collector 里。我們使用 sync.WaitGroup 實(shí)現(xiàn)協(xié)程之間的同步。現(xiàn)在我們有了一個(gè)很好的解決方案，一起來(lái)測(cè)試下。

// main.go 
 
package main 
 
import ( 
 "fmt" 
 "time" 
 
 "github.com/Joker666/goworkerpool/workerpool" 
) 
 
func main() { 
 var allTask []*workerpool.Task 
 for i := 1; i <= 100; i++ { 
  task := workerpool.NewTask(func(data interface{}) error { 
   taskID := data.(int) 
   time.Sleep(100 * time.Millisecond) 
   fmt.Printf("Task %d processed\n", taskID) 
   return nil 
  }, i) 
  allTask = append(allTask, task) 
 } 
 
 pool := workerpool.NewPool(allTask, 5) 
 pool.Run() 
}

上面的代碼，創(chuàng)建了 100 個(gè)任務(wù)并且使用 5 個(gè)并發(fā)處理這些任務(wù)。

輸出如下：

Worker 3 processes task 98 
Task 92 processed 
Worker 2 processes task 99 
Task 98 processed 
Worker 5 processes task 100 
Task 99 processed 
Task 100 processed 
Took ===============> 2.0056295s

處理 100 個(gè)任務(wù)花費(fèi)了 2s，如何我們將并發(fā)數(shù)提高到 10，我們會(huì)看到處理完所有任務(wù)只需要大約 1s。

我們通過(guò)實(shí)現(xiàn) workerPool 構(gòu)建了一個(gè)健壯的解決方案，具有并發(fā)性、錯(cuò)誤處理、數(shù)據(jù)處理等功能。這是個(gè)通用的包，不耦合具體的實(shí)現(xiàn)。我們可以使用它來(lái)解決一些大問(wèn)題。

進(jìn)一步擴(kuò)展：后臺(tái)處理任務(wù)

實(shí)際上，我們還可以進(jìn)一步擴(kuò)展上面的解決方案，以便 worker 可以在后臺(tái)等待我們投遞新的任務(wù)并處理。為此，代碼需要做一些修改，Task 結(jié)構(gòu)體保持不變，但是需要小改下 Worker，看下面代碼：

// workerpool/worker.go 
 
// Worker handles all the work 
type Worker struct { 
 ID       int 
 taskChan chan *Task 
 quit     chan bool 
} 
 
// NewWorker returns new instance of worker 
func NewWorker(channel chan *Task, ID int) *Worker { 
 return &Worker{ 
  ID:       ID, 
  taskChan: channel, 
  quit:     make(chan bool), 
 } 
} 
 
.... 
 
// StartBackground starts the worker in background waiting 
func (wr *Worker) StartBackground() { 
 fmt.Printf("Starting worker %d\n", wr.ID) 
 
 for { 
  select { 
  case task := <-wr.taskChan: 
   process(wr.ID, task) 
  case <-wr.quit: 
   return 
  } 
 } 
} 
 
// Stop quits the worker 
func (wr *Worker) Stop() { 
 fmt.Printf("Closing worker %d\n", wr.ID) 
 go func() { 
  wr.quit <- true 
 }() 
}

Worker 結(jié)構(gòu)體新加 quit channel，并且新加了兩個(gè)方法。StartBackgorund() 在 for 循環(huán)里使用 select-case 從 taskChan 隊(duì)列讀取任務(wù)并處理，如果從 quit 讀取到結(jié)束信號(hào)就立即返回。Stop() 方法負(fù)責(zé)往 quit 寫(xiě)入結(jié)束信號(hào)。

添加完這兩個(gè)新的方法之后，我們來(lái)修改下 Pool：

// workerpool/pool.go 
 
type Pool struct { 
 Tasks   []*Task 
 Workers []*Worker 
 
 concurrency   int 
 collector     chan *Task 
 runBackground chan bool 
 wg            sync.WaitGroup 
} 
 
// AddTask adds a task to the pool 
func (p *Pool) AddTask(task *Task) { 
 p.collector <- task 
} 
 
// RunBackground runs the pool in background 
func (p *Pool) RunBackground() { 
 go func() { 
  for { 
   fmt.Print("⌛ Waiting for tasks to come in ...\n") 
   time.Sleep(10 * time.Second) 
  } 
 }() 
 
 for i := 1; i <= p.concurrency; i++ { 
  worker := NewWorker(p.collector, i) 
  p.Workers = append(p.Workers, worker) 
  go worker.StartBackground() 
 } 
 
 for i := range p.Tasks { 
  p.collector <- p.Tasks[i] 
 } 
 
 p.runBackground = make(chan bool) 
 <-p.runBackground 
} 
 
// Stop stops background workers 
func (p *Pool) Stop() { 
 for i := range p.Workers { 
  p.Workers[i].Stop() 
 } 
 p.runBackground <- true 
}

Pool 結(jié)構(gòu)體添加了兩個(gè)成員：Workers 和 runBackground，Workers 保存所有的 worker，runBackground 用于維持 pool 存活狀態(tài)。

添加了三個(gè)新的方法，AddTask() 方法用于往 collector 添加任務(wù);RunBackground() 方法衍生出一個(gè)無(wú)限運(yùn)行的 goroutine，以便 pool 維持存活狀態(tài)，因?yàn)?runBackground 信道是空，讀取空的 channel 會(huì)阻塞，所以 pool 能維持運(yùn)行狀態(tài)。接著，在協(xié)程里面啟動(dòng) worker;Stop() 方法用于停止 worker，并且給 runBackground 發(fā)送停止信號(hào)以便結(jié)束 RunBackground() 方法。

我們來(lái)看下具體是如何工作的。

如果是在現(xiàn)實(shí)的業(yè)務(wù)場(chǎng)景中，pool 將會(huì)與 HTTP 服務(wù)器一塊運(yùn)行并消耗任務(wù)。我們通過(guò) for 無(wú)限循環(huán)模擬這種這種場(chǎng)景，如果滿足某一條件，pool 將會(huì)停止。

// main.go 
 
... 
 
pool := workerpool.NewPool(allTask, 5) 
go func() { 
 for { 
  taskID := rand.Intn(100) + 20 
 
  if taskID%7 == 0 { 
   pool.Stop() 
  } 
 
  time.Sleep(time.Duration(rand.Intn(5)) * time.Second) 
  task := workerpool.NewTask(func(data interface{}) error { 
   taskID := data.(int) 
   time.Sleep(100 * time.Millisecond) 
   fmt.Printf("Task %d processed\n", taskID) 
   return nil 
  }, taskID) 
  pool.AddTask(task) 
 } 
}() 
pool.RunBackground()

當(dāng)執(zhí)行上面的代碼時(shí)，我們就會(huì)看到有隨機(jī)的 task 被投遞到后臺(tái)運(yùn)行的 workers，其中某一個(gè) worker 會(huì)讀取到任務(wù)并完成處理。當(dāng)滿足某一條件時(shí)，程序便會(huì)停止退出。