errgroup (golang.org/x/sync/errgroup) 為一組協(xié)程 goroutines 在執(zhí)行共同任務(wù)的子任務(wù)時(shí)，提供同步、錯(cuò)誤傳播和上下文取消的功能。

對(duì)于要等待 n 個(gè)線程完成后再進(jìn)行下一步的同步操作的做法，常見使用 sync.WaitGroup 來(lái)等待一組事件。

errgroup.Group 與 sync.WaitGroup 類似，但它增加了對(duì)返回錯(cuò)誤任務(wù)的處理能力，以及限制協(xié)程并發(fā)數(shù)的能力。

使用方法與 WaitGroup 類似，只是封裝了 WaitGroup 的 Add() 和 Wait() 方法，解決 WaitGroup 無(wú)法返回錯(cuò)誤的問(wèn)題。

package main
import (
 "fmt"
 "time"
 "golang.org/x/sync/errgroup"
)
func main() {
 g := &errgroup.Group{}
 for i := 0; i < 5; i++ {
  index := i
  g.Go(func() error {
   fmt.Printf("start to execute the %d gorouting\n", index)
   time.Sleep(time.Duration(index) * time.Second)
   if index%2 == 0 {
    return fmt.Errorf("something has failed on grouting:%d", index)
   }
   fmt.Printf("gorouting:%d end\n", index)
   return nil
  })
 }
 if err := g.Wait(); err != nil {
  fmt.Println(err)
 }
}
// Output:
// start to execute the 4 gorouting
// start to execute the 1 gorouting
// start to execute the 0 gorouting
// start to execute the 2 gorouting
// start to execute the 3 gorouting
// gorouting:1 end
// gorouting:3 end
// something has failed on grouting:0

如果多個(gè) goroutine 出現(xiàn)錯(cuò)誤，errgroup 只會(huì)獲取到第一個(gè)出錯(cuò)的 goroutine 的錯(cuò)誤信息。不管是否有協(xié)程執(zhí)行失敗，wait() 都要等待所有協(xié)程執(zhí)行完成。

支持 context ：

g, _ := errgroup.WithContext(context.Background()) // 支持 context

Wait() 方法可多次調(diào)用，依然可以得到 group 的 error 信息：

 ...
 if err := g.Wait(); err != nil {
  fmt.Println(err)
 }
 if err := g.Wait(); err != nil { // 可再次調(diào)用 Wait，依然可以得到 group 的 error 信息
  fmt.Println(err)
 }

限制最大并發(fā)數(shù)：

SetLimit() 方法用于限制該組中最多同時(shí)運(yùn)行的 goroutine 數(shù)量，參數(shù)代表的是當(dāng)前同時(shí)處于活動(dòng)狀態(tài)(處理業(yè)務(wù))的 goroutine 的最多數(shù)量。

package main
import (
 "log"
 "time"
 "golang.org/x/sync/errgroup"
)
func main() {
 jobs := make(chan int, 10)
 go func() {
  for i := 0; i < 8; i++ {
   jobs <- i + 1
  }
  close(jobs)
 }()
 eg:= &errgroup.Group{}
 eg.SetLimit(3)
 for j := range jobs {
  j := j
  eg.Go(func() error {
   log.Printf("handle job: %d\n", j)
   time.Sleep(2 * time.Second)
   return nil
  })
 }
 eg.Wait()
}

上面的示例創(chuàng)建了一組 goroutines 來(lái)處理 job，同一時(shí)間允許最多 3 個(gè) goroutine 處于活動(dòng)狀態(tài)。

Output:

2024/12/17 18:28:19 handle job: 3
2024/12/17 18:28:19 handle job: 1
2024/12/17 18:28:19 handle job: 2
2024/12/17 18:28:21 handle job: 4
2024/12/17 18:28:21 handle job: 6
2024/12/17 18:28:21 handle job: 5
2024/12/17 18:28:23 handle job: 7
2024/12/17 18:28:23 handle job: 8

從示例運(yùn)行結(jié)果中的時(shí)間戳我們可以看到：雖然我們創(chuàng)建了很多 goroutine，但同一時(shí)間內(nèi)處理活動(dòng)狀態(tài)(正在處理 job)的 goroutine 的數(shù)量最多為 3 個(gè)。

其內(nèi)部實(shí)現(xiàn)就是將帶緩沖 channel 用作計(jì)數(shù)信號(hào)量 (counting semaphore) 來(lái)限制最大并發(fā)數(shù)。

到此這篇關(guān)于Go 并發(fā)控制 errgroup.Group的實(shí)現(xiàn)示例的文章就介紹到這了,更多相關(guān)Go errgroup.Group內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

最新評(píng)論