Go語(yǔ)言動(dòng)態(tài)并發(fā)控制sync.WaitGroup的靈活運(yùn)用示例詳解

更新時(shí)間：2023年11月22日 11:42:29 作者：Go先鋒

本文將講解 sync.WaitGroup 的使用方法、原理以及在實(shí)際項(xiàng)目中的應(yīng)用場(chǎng)景,用清晰的代碼示例和詳細(xì)的注釋,助力讀者掌握并發(fā)編程中等待組的使用技巧

概述

在并發(fā)編程中，控制主程序等待所有 Goroutine 完成任務(wù)是一項(xiàng)關(guān)鍵任務(wù)。Go 語(yǔ)言提供了 sync.WaitGroup 來(lái)解決這一問(wèn)題。

1. 基本使用

1.1 初始化和添加計(jì)數(shù)

package main
import (
  "fmt"
  "sync"
  "time"
)
func main() {
  var wg sync.WaitGroup
  for i := 1; i <= 3; i++ {
    wg.Add(1)
    go worker(i, &wg)
  }
  wg.Wait()
  fmt.Println("All workers have completed.")
}
func worker(id int, wg *sync.WaitGroup) {
  defer wg.Done()
  fmt.Printf("Worker %d started\n", id)
  time.Sleep(2 * time.Second)
  fmt.Printf("Worker %d completed\n", id)
}

在上面示例中，用一個(gè) sync.WaitGroup 實(shí)例 wg，然后使用 wg.Add(1) 來(lái)增加計(jì)數(shù)，表示有一個(gè) Goroutine 需要等待。

在每個(gè) Goroutine 的結(jié)束處，使用 defer wg.Done() 來(lái)減少計(jì)數(shù)，表示一個(gè) Goroutine 已完成。

最后，用 wg.Wait() 來(lái)等待所有 Goroutine 完成。

1.2 處理錯(cuò)誤

package main
import (
  "fmt"
  "sync"
  "time"
)
func main() {
  var wg sync.WaitGroup
  for i := 1; i <= 3; i++ {
    wg.Add(1)
    go workerWithError(i, &wg)
  }
  wg.Wait()
  fmt.Println("All workers have completed.")
}
func workerWithError(id int, wg *sync.WaitGroup) {
  defer wg.Done()
  fmt.Printf("Worker %d started\n", id)
  time.Sleep(2 * time.Second)
  // 模擬錯(cuò)誤發(fā)生
  if id == 2 {
    fmt.Printf("Worker %d encountered an error\n", id)
    return
  }
  fmt.Printf("Worker %d completed\n", id)
}

有時(shí)候，需要在 Goroutine 中處理錯(cuò)誤。在這個(gè)示例中，當(dāng) id 為 2 時(shí)，模擬了一個(gè)錯(cuò)誤的情況。

通過(guò)在錯(cuò)誤發(fā)生時(shí)提前返回，可以確保計(jì)數(shù)正確減少，避免等待組出現(xiàn)死鎖。

2. 多級(jí)等待組

2.1 嵌套使用

package main
import (
  "fmt"
  "sync"
  "time"
)
func main() {
  var outerWG sync.WaitGroup
  var innerWG sync.WaitGroup
  for i := 1; i <= 2; i++ {
    outerWG.Add(1)
    go outerWorker(i, &outerWG, &innerWG)
  }
  outerWG.Wait()
  fmt.Println("All outer workers have completed.")
}
func outerWorker(id int, outerWG, innerWG *sync.WaitGroup) {
  defer outerWG.Done()
  fmt.Printf("Outer Worker %d started\n", id)
  for j := 1; j <= 3; j++ {
    innerWG.Add(1)
    go innerWorker(id, j, innerWG)
  }
  innerWG.Wait()
  fmt.Printf("Outer Worker %d completed\n", id)
}
func innerWorker(outerID, innerID int, wg *sync.WaitGroup) {
  defer wg.Done()
  fmt.Printf("Inner Worker %d of Outer Worker %d started\n", innerID, outerID)
  time.Sleep(2 * time.Second)
  fmt.Printf("Inner Worker %d of Outer Worker %d completed\n", innerID, outerID)
}

在示例中，使用了嵌套的 sync.WaitGroup。

外部的等待組 outerWG 等待所有外部 Goroutine 完成，而每個(gè)外部 Goroutine 內(nèi)部的 innerWG 則等待其內(nèi)部的所有 Goroutine 完成。

2.2 動(dòng)態(tài)添加等待組

package main
import (
  "fmt"
  "sync"
  "time"
)
func main() {
  var dynamicWG sync.WaitGroup
  for i := 1; i <= 3; i++ {
    dynamicWG.Add(1)
    go dynamicWorker(i, &dynamicWG)
  }
  // 模擬動(dòng)態(tài)添加更多任務(wù)
  time.Sleep(1 * time.Second)
  for i := 4; i <= 6; i++ {
    dynamicWG.Add(1)
    go dynamicWorker(i, &dynamicWG)
  }
  dynamicWG.Wait()
  fmt.Println("All dynamic workers have completed.")
}
func dynamicWorker(id int, wg *sync.WaitGroup) {
  defer wg.Done()
  fmt.Printf("Dynamic Worker %d started\n", id)
  time.Sleep(2 * time.Second)
  fmt.Printf("Dynamic Worker %d completed\n", id)
}

在上述示例中，創(chuàng)建了一個(gè)等待組 dynamicWG，然后在運(yùn)行時(shí)動(dòng)態(tài)添加了更多的任務(wù)。

用這種方式，可以動(dòng)態(tài)地管理需要等待的 Goroutine 數(shù)量。

3. 超時(shí)處理

3.1 帶超時(shí)的等待

package main
import (
  "fmt"
  "sync"
  "time"
)
func main() {
  var timeoutWG sync.WaitGroup
  for i := 1; i <= 3; i++ {
    timeoutWG.Add(1)
    go timeoutWorker(i, &timeoutWG)
  }
  // 等待最多5秒，超時(shí)則不再等待
  timeout := time.After(5 * time.Second)
  done := make(chan struct{})
  go func() {
    timeoutWG.Wait()
    close(done)
  }()
  select {
  case <-done:
    fmt.Println("All timeout workers have completed.")
  case <-timeout:
    fmt.Println("Timeout reached. Not all workers have completed.")
  }
}
func timeoutWorker(id int, wg *sync.WaitGroup) {
  defer wg.Done()
  fmt.Printf("Timeout Worker %d started\n", id)
  time.Sleep(time.Duration(id) * time.Second)
  fmt.Printf("Timeout Worker %d completed\n", id)
}

在上面示例中，用 time.After 創(chuàng)建了一個(gè) 5 秒的超時(shí)通道。

在另一個(gè) Goroutine 中監(jiān)聽(tīng)等待組的完成情況，可以在超時(shí)或任務(wù)完成時(shí)得知等待的最終結(jié)果。

3.2 處理超時(shí)錯(cuò)誤

package main
import (
  "errors"
  "fmt"
  "sync"
  "time"
)
func main() {
  var timeoutWG sync.WaitGroup
  for i := 1; i <= 3; i++ {
    timeoutWG.Add(1)
    go timeoutWorkerWithError(i, &timeoutWG)
  }
  // 等待最多5秒，超時(shí)則返回錯(cuò)誤
  err := waitWithTimeout(&timeoutWG, 5*time.Second)
  if err != nil {
    fmt.Printf("Timeout reached. Not all workers have completed. Error: %v\n", err)
  } else {
    fmt.Println("All timeout workers have completed.")
  }
}
func timeoutWorkerWithError(id int, wg *sync.WaitGroup) {
  defer wg.Done()
  fmt.Printf("Timeout Worker %d started\n", id)
  time.Sleep(time.Duration(id) * time.Second)
  // 模擬錯(cuò)誤發(fā)生
  if id == 2 {
    fmt.Printf("Timeout Worker %d encountered an error\n", id)
    return
  }
  fmt.Printf("Timeout Worker %d completed\n", id)
}
func waitWithTimeout(wg *sync.WaitGroup, timeout time.Duration) error {
  done := make(chan struct{})
  go func() {
    defer close(done)
    wg.Wait()
  }()
  select {
  case <-done:
    return nil
  case <-time.After(timeout):
    return errors.New("timeout reached")
  }
}

有時(shí)候，希望在程序超時(shí)的時(shí)候返回一個(gè)錯(cuò)誤。

在這個(gè)示例中，用封裝等待組的超時(shí)檢查，可以在主程序中獲得一個(gè)清晰的錯(cuò)誤提示。

總結(jié)

通過(guò)討論 sync.WaitGroup 的基本用法、避免常見(jiàn)錯(cuò)誤以及實(shí)際應(yīng)用，深入了解了這個(gè)強(qiáng)大的同步工具。

在 Go 語(yǔ)言并發(fā)編程中，合理使用 sync.WaitGroup 能夠優(yōu)雅地處理并發(fā)等待，確保主程序在所有任務(wù)完成后再繼續(xù)執(zhí)行。

以上就是Go語(yǔ)言動(dòng)態(tài)并發(fā)控制sync.WaitGroup的靈活運(yùn)用示例詳解的詳細(xì)內(nèi)容，更多關(guān)于Go sync.WaitGroup動(dòng)態(tài)并發(fā)控制的資料請(qǐng)關(guān)注腳本之家其它相關(guān)文章！

您可能感興趣的文章: