同步協程的必備工具: WaitGroup

来源:https://www.cnblogs.com/chenjiazhan/archive/2023/03/17/17228255.html
-Advertisement-
Play Games

類載入器 類載入的過程 類載入器的功能 將.class文件【物理文件：在硬碟中】載入到Java虛擬機的記憶體中【搬用工】。 類載入的時機情況分析： //1，當創建Fu對象的時候【Fu.class會被載入到Java虛擬機】 Fu f = new Fu(); //2,調用類的靜態方法【Fu.class會被 ...


1. 簡介

本文將介紹 Go 語言中的 WaitGroup 併發原語，包括 WaitGroup 的基本使用方法、實現原理、使用註意事項以及常見的使用方式。能夠更好地理解和應用 WaitGroup 來協調多個 Goroutine 的執行，提高 Go 併發編程的效率和穩定性。

2. 基本使用

2.1 定義

WaitGroup是Go語言標準庫中的一個結構體，它提供了一種簡單的機制，用於同步多個協程的執行。適用於需要併發執行多個任務並等待它們全部完成後才能繼續執行後續操作的場景。

2.2 使用方式

首先主協程創建WaitGroup實例，然後在每個協程的開始處，調用Add(1)方法，表示需要等待一個任務執行完成，然後協程在任務執行完成之後，調用Done方法，表示任務已經執行完成了。

主協程中，需要調用Wait()方法，等待所有協程完成任務，示例如下:

func main(){
    //首先主協程創建WaitGroup實例
    var wg sync.WaitGroup
    // 開始時調用Add方法表示有個任務開始執行
    wg.Add(1)
    go func() {
        // 開始執行...
        //完成之後,調用Done方法
        wg.Done()
    }()
    // 調用Wait()方法，等待所有協程完成任務
    wg.Wait()
    // 執行後續邏輯
}

2.3 使用例子

package main

import (
    "fmt"
    "sync"
)

func main() {
    var wg sync.WaitGroup
    for i := 0; i < 5; i++ {
       wg.Add(1)
       go func(i int) {
          defer wg.Done()
          fmt.Printf("任務%d開始執行\n", i)
          // 模擬協程任務執行一段時間
          time.Sleep(time.Duration(rand.Int() % 100))
          // 線程任務執行完成
          fmt.Printf("任務%d執行完畢\n", i)
       }(i)
    }
    fmt.Println("主協程開始等待所有任務執行完成...")
    wg.Wait()
    fmt.Println("所有協程已經執行完畢...")
}

在這個例子中，我們使用了sync.WaitGroup來等待5個協程執行完畢。在迴圈中，每創建一個任務，我們調用一次wg.Add(1)方法，然後啟動一個協程去執行任務，當協程完成任務後，調用wg.Done方法，告知主協程任務已經執行完畢。然後主協程會在5個協程任務全部執行完畢之後，才會繼續向下執行。

3.實現原理

3.1 設計初衷

WaitGroup的設計初衷就是為了等待一組操作完成後再執行下一步操作，通常會在一組協程中使用。

3.2 實現

sync.WaitGroup 結構體中的 state1 和 state2 欄位是用於實現 WaitGroup 功能的重要變數。

type WaitGroup struct {
   noCopy noCopy

   state1 uint64
   state2 uint32
}

由於 WaitGroup 需要等待一組操作完成之後再執行，因此需要等待所有操作完成之後才能繼續執行。為了實現這個功能，WaitGroup 使用了一個計數器 counter 來記錄還有多少個操作沒有完成，如果 counter 的值為 0，則表示所有操作已經完成。

同時，WaitGroup 在所有任務都完成之後，需要喚醒所有處於等待的協程，此時需要知道有多少個協程處於等待狀態。為了實現這個功能，WaitGroup 使用了一個等待計數器 waiter 來記錄當前有多少個協程正在等待操作完成。

這裡WaitGroup對於計數器和等待計數器的實現，是通過一個64位無符號整數來實現的，也就是WaitGroup結構體中的state1，其中高32位保存了任務計數器counter的值，低32位保存了等待計數器waiter的值。當我們創建一個 WaitGroup 實例時，該實例的任務計數器和等待計數器都被初始化為 0。

而且，等待協程需要等待所有任務完成之後才能繼續執行，所以等待協程在任務未完成時會被阻塞，當任務全部完成後，自動被喚醒。WaitGroup使用 state2 用於實現信號量機制。通過調用 runtime_Semacquire() 和 runtime_Semrelease() 函數，可以在不阻塞線程的情況下進行等待和通知操作。下麵是Add，Done和Wait方法的具體實現:

調用 Add() 方法增加/減小counter的值，delta的值可以是正數，也可以是負數，下麵是Add方法的源碼實現:

func (wg *WaitGroup) Add(delta int) {
   // delta 的值可以為負數,Done方法便是通過Add(-1)來實現的
   // statep: 為state1的地址  semap: 為state2的地址
   statep, semap := wg.state()
   // 高32位的值 加上 delta,增加任務計數器的值
   state := atomic.AddUint64(statep, uint64(delta)<<32)
   // v: 取高32位數據,獲取到待完成任務數
   v := int32(state >> 32)
   // 取低32位數據,獲取到等待線程的值
   w := uint32(state)
   // v > 0: 說明還有待完成的任務數,此時不應該喚醒等待協程
   // w = 0: 說明沒有協程在等待,此時可以直接退出
   if v > 0 || w == 0 {
      return
   }     
   // 此時v = 0,所有任務都完成了,喚醒等待協程
   *statep = 0
   for ; w != 0; w-- {
      runtime_Semrelease(semap, false, 0)
   }
}

調用 Done() 方法表示完成了一個任務，通過調用Add方法，delta值為-1，減少任務計數器counter的值，當其歸為0時，便自動喚醒所有處於等待的協程。

// Done decrements the WaitGroup counter by one.
func (wg *WaitGroup) Done() {
   wg.Add(-1)
}

調用Wait方法，等待任務執行完成，增加等待計數器Waiter的值:

func (wg *WaitGroup) Wait() {
   // statep: 為state1的地址  semap: 為state2的地址
   statep, semap := wg.state()
   for {
      // 載入state1的值
      state := atomic.LoadUint64(statep)
      // v: 取高32位數據,獲取到待完成任務數
      v := int32(state >> 32)
      // 沒有任務待執行,全部都完成了
      if v == 0 {
         return
      }
      // 增加waiter計數器的值
      if atomic.CompareAndSwapUint64(statep, state, state+1) {
          // 等待被喚醒
         runtime_Semacquire(semap)
         return
      }
   }
}

3.3 實現補充

Add方法，Done方法以及Wait方法實現中，有一些異常場景的驗證邏輯被我刪除掉了。當出現異常場景時，說明用戶使用方式和WaitGroup的設計初衷相違背了，此時WaitGroup就會直接panic。

下麵通過說明使用的註意事項，來間接介紹WaitGroup的異常驗證邏輯。

4.使用註意事項

4.1 Add方法和Done方法需要成對出現

下麵是一個Add方法和Done方法沒有成對出現的例子，此時Add方法調多了，此時計數器永遠大於0，Wait 方法會一直阻塞等待。

package main

import (
    "fmt"
    "sync"
)

func main() {
    var wg sync.WaitGroup

    wg.Add(2)

    go func() {
        defer wg.Done()
        fmt.Println("Goroutine 1")
    }()

    go func() {
        fmt.Println("Goroutine 2")
    }()

    wg.Wait()

    fmt.Println("All goroutines finished")
}

在上述代碼中，我們調用了wg.Add(2)，但只調用了一次wg.Done()。這會導致counter的值大於0，因此調用wg.Wait()會被永久阻塞，不會繼續向下繼續執行。

還有另外一種情況時Done方法調用多了，此時任務計數器counter的值為負數，從WaitGroup設計的語意來看，就是需要等待完成的任務數為負數，這個不符合預期，此時將會直接panic

package main

import (
    "fmt"
    "sync"
)

func main() {
    var wg sync.WaitGroup
    wg.Add(1)

    go func() {
        fmt.Println("Goroutine 1 started")
        wg.Done() // 第一次調用Done方法
        wg.Done() // 第二次調用Done方法
        fmt.Println("Goroutine 1 completed")
    }()

    wg.Wait()
    fmt.Println("All goroutines completed")
}

在上面的例子中，我們啟動了一個goroutine，第一次調用Add方法，counter的值變為1，在第14行調用Done，此時計數器的值變為0，此時等待中的goroutine將會被喚醒。在第15行又調用了一次Done方法，當counter減小為0時，再次調用Done方法會導致panic，因為此時waitGroup的計數器已經為0，再次減少將導致負數計數，這是不被允許的。

所以在調用Done方法時，需要保證每次調用都與Add方法的調用一一對應，否則會導致程式出現錯誤。

4.2 在所有任務都已經添加之後，才調用Wait方法進行等待

WaitGroup的設計初衷就是為了等待一組操作完成後再執行下一步操作。所以，如果在所有任務添加之前，便調用Wait方法進行等待，此時有可能會導致等待協程提前被喚醒，執行下一步操作，而尚未添加的任務則不會被等待，這違反了WaitGroup的設計初衷，也不符合預期。下麵是一個簡單的例子:

package main

import (
        "fmt"
        "sync"
        "time"
)

func main() {
        var wg sync.WaitGroup
        for i := 1; i <= 3; i++ {
           go func(id int) {
              wg.Add(1)
              defer wg.Done()
              fmt.Printf("Goroutine %d started\n", id)
              time.Sleep(time.Duration(id) * time.Second) 
              fmt.Printf("Goroutine %d finished\n", id)
           }(i)
        }
        
        // 不等待所有任務添加，就開始等待
        wg.Wait()
        fmt.Println("All goroutines finished")
        time.Sleep(10 * time.Second)
}

代碼執行結果如下，等待協程被提前喚醒，執行之後的操作，而子任務在等待協程喚醒後才開始執行:

All goroutines finished
Goroutine 1 started
Goroutine 3 started
Goroutine 2 started
Goroutine 1 finished
Goroutine 2 finished
Goroutine 3 finished

在這個例子中，我們創建了三個協程並列印出它們開始和結束的消息。但是，我們沒有在任務開始前調用Add方法添加任務，而是在任務開始之後再調用Add方法添加任務。

這可能會導致某些任務未被加入到WaitGroup中，等待協程就調用了wg.Wait方法，這樣就會導致一些任務未被加入WaitGrou，從而導致等待協程不會等待這些任務執行完成。如果這種情況發生了，我們會看到"All goroutines finished"被輸出，但實際上有一些協程還沒有完成。

因此，我們應該在所有任務添加完畢之後再調用Wait方法，以保證等待的正確性。

5. WaitGroup常見使用方式

在函數或方法中使用，如果一個大任務可以拆分為多個獨立的子任務，此時會將其進行拆分，並使用多個協程來併發執行這些任務，提高執行效率，同時使用WaitGroup等待所有子任務執行完成，完成協程間的同步。

使用方式也比較簡單，先創建一個 sync.WaitGroup，在函數/方法中啟動多個協程，每個協程執行一個任務。然後在協程開始執行任務前，調用 WaitGroup.Add(1)，表示有一個任務要執行，然後在任務執行完成後調用 WaitGroup.Done()，表示這個任務執行完成了。

最後，在函數/方法返回之前，需要調用 WaitGroup.Wait()，等待所有的任務執行完成。大概示例如下:

func funcName() {
    var wg sync.WaitGroup
    for _, 任務 := range 任務列表 {
      wg.Add(1)
      go func() {
        defer wg.Done()
        //執行任務
      }
   }
   // 調用wait方法等待所有任務完成
   wg.Wait()
}

下麵來看go-redis中ClusterClient結構體中ForEachMaster方法中對於WaitGroup的使用。ForEachMaster方法通常用於在 Redis 集群中執行針對所有主節點的某種操作，例如在集群中添加或刪除鍵，或者執行一些全局的診斷操作，具體執行的操作由傳入參數fn指定。

這裡ForEachMaster方法會對所有主節點執行某種操作，這裡的實現是對所有主節點執行某種操作這個大任務，拆分為多個獨立的子任務，每個子任務完成對一個Master節點執行指定操作，然後每個子任務啟動一個協程去執行，主協程使用WaitGroup等待所有協程完成指定子任務，ForEachMaster也就完成了對所有主節點執行某種操作的任務。具體實現如下：

func (c *ClusterClient) ForEachMaster(
   ctx context.Context,
   fn func(ctx context.Context, client *Client) error,
) error {
   // 重新載入集群狀態,以確保狀態信息是最新的
   state, err := c.state.ReloadOrGet(ctx)
   if err != nil {
      return err
   }
   var wg sync.WaitGroup
   // 用於協程間通信
   errCh := make(chan error, 1)
    // 獲取到redis集群中所有的master節點
   for _, master := range state.Masters {
      // 啟動一個協程來執行該任務
      wg.Add(1)
      go func(node *clusterNode) {
         // 任務完成時,調用Done告知WaitGroup任務已完成
         defer wg.Done()
         err := fn(ctx, node.Client)
         if err != nil {
            select {
            case errCh <- err:
            default:
            }
         }
      }(master)
   }
   // 主協程等待所有任務完成
   wg.Wait()
   return nil
 }

6.總結

本文介紹了 Go 語言中的 WaitGroup 併發原語，它提供了一種簡單且強大的機制來協調多個 Goroutine 的執行。我們首先學習了 WaitGroup 的基本使用方法，包括如何創建 WaitGroup、如何向計數器中添加值、如何等待所有 Goroutine 完成以及如何在 Goroutine 中通知 WaitGroup 完成。

接著，我們瞭解了 WaitGroup 的實現原理，包括計數器和等待計數器的實現。瞭解了實現原理之後，我們可以更好地理解 WaitGroup 的內部機制以及如何更好地使用它來實現我們的需求。

在接下來的部分中，我們介紹了一些使用 WaitGroup 的註意事項，以及常見的使用方式。基於此，我們完成了對WaitGroup的介紹。


您的分享是我們最大的動力!

-Advertisement-
Play Games
更多相關文章
  • 萬字血書Vue—Vue語法
    模板語法 插值語法 Mustache插值採用{{ }}，用於解析標簽體內容，將Vue實例中的數據插入DOM中 <h1>Hello {{name}}</h1> 指令語法 指令用於解析標簽，是vue為開發者提供的一套模板語法，輔助開發者渲染頁面的基本結構。 （指令是vue開發中最基礎、最常用、最簡單的知 ...
  • uni-app 實現輪播圖組件父容器背景色隨圖片主題色改變
    uni-app 實現輪播圖組件父容器背景色隨圖片主題色改變 實現思路 1、獲取輪播圖主題色，通過 canvas 獲取圖片主題色。 2、隨著輪播圖組件圖片的輪播，動態設置父容器背景色為圖片的主題色。 實現代碼 <template> <view class="container"> <canvas ca ...
  • 萬字血書Vue—走近Vue
    Vue是什麼？ Vue是一套用於構建用戶界面的漸進式JavaScript框架 構建用戶界面：用vue往html頁面中填充數據 漸進式：Vue可以自底向上逐層的應用，從輕量小巧核心庫的簡單應用，到引入各式各樣插件的複雜應用。 框架：一整套現成的解決方案，遵守框架的規範，學習框架，就是學習框架中的規定用 ...
  • 前端設計模式——組合模式
    組合模式（Composite Pattern）是一種結構型設計模式，它允許將對象組合成樹形結構，並且可以像操作單個對象一樣操作整個樹形結構。 組合模式的核心思想是將對象組織成樹形結構，其中包含組合對象和葉子對象兩種類型。組合對象可以包含葉子對象或其他組合對象，從而形成一個樹形結構。 組合模式可以應用 ...
  • ChatGPT 虛擬號碼：手機號碼，簡訊驗證碼接碼推薦
    ChatGPT 虛擬號碼是什麼？ 如何使用虛擬號註冊 ChatGPT，用來收手機驗證碼呢？先瞭解它是什麼 虛擬號碼是一種虛擬電話號碼，它可以用於接收和發送簡訊，但不會顯示真實的號碼。它可以用於保護用戶的隱私，也可以用於接收垃圾簡訊。 ChatGPT 虛擬號碼簡訊驗證碼接碼平臺 有幾個網站提供與 ht ...
  • 【java筆記】cn.hutool.core.io.IORuntimeException Not a file。。。
    報錯信息 如題， cn.hutool.core.io.IORuntimeException: Not a file.... 報錯位置 FileReader reader = new FileReader(path); 初步分析 檢查下來發現，path實際對應的是一個文件夾，而不是文件。 文件來源關鍵 ...
  • TP 判斷IP是否在國內
    環境ThinkPHP+Redis 1.IP保存文件，文件名自定義，與後文對應 2.獲取IP信息腳本.sh文件 #!/bin/bash #variables ip_txt_path=/www/wwwroot/checkip/china_ip.txt; ip_url='http://ftp.apnic. ...
  • 使用python爬取豆瓣電影短評評論內容
    需求：爬取豆瓣電影短評評論文本內容 目標：將爬取的文本存入 excel 中 爬蟲步驟： 1.拼接分頁網址，迴圈請求分頁數據，獲取HTML代碼 2.分析獲取到的HTML代碼，解析出所需要的數據，提取內容 3.存儲爬取到的數據 準備工作： 1.開發工具 pycharm 2.模塊 requests、bs4 ...
一周排行
    -Advertisement-
    Play Games
  • 移動開發（一）：使用.NET MAUI開發第一個安卓APP
    移動開發（一）：使用.NET MAUI開發第一個安卓APP 對於工作多年的C#程式員來說，近來想嘗試開發一款安卓APP，考慮了很久最終選擇使用.NET MAUI這個微軟官方的框架來嘗試體驗開發安卓APP，畢竟是使用Visual Studio開發工具，使用起來也比較的順手，結合微軟官方的教程進行了安卓 ...
  • wpf ToggleButton選中效果和一個登錄界面
    前言 QuestPDF 是一個開源 .NET 庫，用於生成 PDF 文檔。使用了C# Fluent API方式可簡化開發、減少錯誤並提高工作效率。利用它可以輕鬆生成 PDF 報告、發票、導出文件等。 項目介紹 QuestPDF 是一個革命性的開源 .NET 庫，它徹底改變了我們生成 PDF 文檔的方 ...
  • 彈幕樹洞項目功能新增篇
    項目地址 項目後端地址： https://github.com/ZyPLJ/ZYTteeHole 項目前端頁面地址： ZyPLJ/TreeHoleVue (github.com) https://github.com/ZyPLJ/TreeHoleVue 目前項目測試訪問地址： http://tree ...
  • 第27篇 sqlserver2022詳細安裝步驟
    話不多說，直接開乾 一.下載 1.官方鏈接下載： https://www.microsoft.com/zh-cn/sql-server/sql-server-downloads 2.在下載目錄中找到下麵這個小的安裝包 SQL2022-SSEI-Dev.exe，運行開始下載SQL server； 二. ...
  • .NET 開源高性能 MQTT 類庫
    前言 隨著物聯網（IoT）技術的迅猛發展，MQTT（消息隊列遙測傳輸）協議憑藉其輕量級和高效性，已成為眾多物聯網應用的首選通信標準。 MQTTnet 作為一個高性能的 .NET 開源庫，為 .NET 平臺上的 MQTT 客戶端與伺服器開發提供了強大的支持。 本文將全面介紹 MQTTnet 的核心功能 ...
  • Serilog文檔翻譯系列（六） - 可用的接收器、增強器、格式化輸出
    Serilog支持多種接收器用於日誌存儲，增強器用於添加屬性，LogContext管理動態屬性，支持多種輸出格式包括純文本、JSON及ExpressionTemplate。還提供了自定義格式化選項，適用於不同需求。 ...
  • 警惕 Visual Studio 屬性求值副作用導致邏輯不符合預期
    目錄簡介獲取 HTML 文檔解析 HTML 文檔測試參考文章 簡介 動態內容網站使用 JavaScript 腳本動態檢索和渲染數據，爬取信息時需要模擬瀏覽器行為，否則獲取到的源碼基本是空的。 本文使用的爬取步驟如下： 使用 Selenium 獲取渲染後的 HTML 文檔 使用 HtmlAgility ...
  • [使用目前最新版]HybridCLR6.9.0+YooAsset2.2.4實現純C# Unity熱更新方案 (一)
    1.前言 什麼是熱更新 游戲或者軟體更新時，無需重新下載客戶端進行安裝，而是在應用程式啟動的情況下，在內部進行資源或者代碼更新 Unity目前常用熱更新解決方案 HybridCLR，Xlua，ILRuntime等 Unity目前常用資源管理解決方案 AssetBundles，Addressable， ...
  • 在 ASP.NET Core Web API 中使用操作篩選器統一處理通用操作
    本文章主要是在C# ASP.NET Core Web API框架實現向手機發送驗證碼簡訊功能。這裡我選擇是一個互億無線簡訊驗證碼平臺，其實像阿裡雲，騰訊雲上面也可以。 首先我們先去 互億無線 https://www.ihuyi.com/api/sms.html 去註冊一個賬號 註冊完成賬號後，它會送 ...
  • 第28篇 如何.net中實現高效可靠數據同步api
    通過以下方式可以高效，並保證數據同步的可靠性 1.API設計 使用RESTful設計，確保API端點明確，並使用適當的HTTP方法（如POST用於創建，PUT用於更新）。 設計清晰的請求和響應模型，以確保客戶端能夠理解預期格式。 2.數據驗證 在伺服器端進行嚴格的數據驗證，確保接收到的數據符合預期格 ...
所有分類