淺談errgroup的使用以及源碼分析

来源:https://www.cnblogs.com/huageyiyangdewo/archive/2023/04/26/17357767.html
-Advertisement-
Play Games

淄博燒烤爆紅出了圈，當你坐在八大局的燒烤攤，面前是火爐、烤串、小餅和蘸料，音樂響起，啤酒倒滿，燒烤靈魂的party即將開場的時候，你系統中的Scheduler（調試器），也自動根據設定的Trigger（觸發器），從容優雅的啟動了一系列的Job（後臺定時任務）。工作一切早有安排，又何須費心勞神呢？因為 ...


本文講解的是golang.org/x/sync這個包中的errgroup

1、errgroup 的基礎介紹

學習過 Go 的朋友都知道 Go 實現併發編程是比較容易的事情，只需要使用go關鍵字就可以開啟一個 goroutine。那對於併發場景中，如何實現goroutine的協調控制呢？常見的一種方式是使用sync.WaitGroup 來進行協調控制。

使用過sync.WaitGroup 的朋友知道，sync.WaitGroup 雖然可以實現協調控制，但是不能傳遞錯誤，那該如何解決呢？聰明的你可能馬上想到使用 chan 或者是 context來傳遞錯誤，確實是可以的。那接下來，我們一起看看官方是怎麼實現上面的需求的呢？

1.1 errgroup的安裝

安裝命令：

go get golang.org/x/sync

//下麵的案例是基於v0.1.0 演示的
go get golang.org/x/[email protected]

1.2 errgroup的基礎例子

這裡我們需要請求3個url來獲取數據，假設請求url2時報錯，url3耗時比較久，需要等一秒。

package main

import (
	"errors"
	"fmt"
	"golang.org/x/sync/errgroup"
	"strings"
	"time"
)

func main()  {
	queryUrls := map[string]string{
		"url1": "http://localhost/url1",
		"url2": "http://localhost/url2",
		"url3": "http://localhost/url3",
	}

	var eg errgroup.Group
	var results []string

	for _, url := range queryUrls {
		url := url
		eg.Go(func() error {
			result, err := query(url)
			if err != nil {
				return err
			}
			results = append(results, fmt.Sprintf("url:%s -- ret: %v", url, result))
			return nil
		})
	}
	
  // group 的wait方法，等待上面的 eg.Go 的協程執行完成，並且可以接受錯誤
	err := eg.Wait()
	if err != nil {
		fmt.Println("eg.Wait error:", err)
		return
	}

	for k, v := range results {
		fmt.Printf("%v ---> %v\n", k, v)
	}
}

func query(url string) (ret string, err error) {
	// 假設這裡是發送請求，獲取數據
	if strings.Contains(url, "url2") {
		// 假設請求 url2 時出現錯誤
		fmt.Printf("請求 %s 中....\n", url)
		return "", errors.New("請求超時")
	} else if strings.Contains(url, "url3") {
		// 假設 請求 url3 需要1秒
		time.Sleep(time.Second*1)
	}
	fmt.Printf("請求 %s 中....\n", url)
	return "success", nil
}

執行結果：

請求 http://localhost/url2 中....
請求 http://localhost/url1 中....
請求 http://localhost/url3 中....
eg.Wait error: 請求超時

果然，當其中一個goroutine出現錯誤時，會把goroutine中的錯誤傳遞出來。

我們自己運行一下上面的代碼就會發現這樣一個問題，請求 url2 出錯了，但是依舊在請求 url3 。因為我們需要聚合 url1、url2、url3 的結果，所以當其中一個出現問題時，我們是可以做一個優化的，就是當其中一個出現錯誤時，取消還在執行的任務，直接返回結果，不用等待任務執行結果。

那應該如何做呢？

這裡假設 url1 執行1秒，url2 執行報錯，url3執行3秒。所以當url2報錯後，就不用等url3執行結束就可以返回了。

package main

import (
	"context"
	"errors"
	"fmt"
	"golang.org/x/sync/errgroup"
	"strings"
	"time"
)

func main()  {
	queryUrls := map[string]string{
		"url1": "http://localhost/url1",
		"url2": "http://localhost/url2",
		"url3": "http://localhost/url3",
	}

	var results []string
	ctx, cancel := context.WithCancel(context.Background())
	eg, errCtx := errgroup.WithContext(ctx)

	for _, url := range queryUrls {
		url := url
		eg.Go(func() error {
			result, err := query(errCtx, url)
			if err != nil {
        //其實這裡不用手動取消，看完源碼就知道為啥了
				cancel()
				return err
			}
			results = append(results, fmt.Sprintf("url:%s -- ret: %v", url, result))
			return nil
		})
	}

	err := eg.Wait()
	if err != nil {
		fmt.Println("eg.Wait error:", err)
		return
	}

	for k, v := range results {
		fmt.Printf("%v ---> %v\n", k, v)
	}
}


func query(errCtx context.Context, url string) (ret string, err error) {
	fmt.Printf("請求 %s 開始....\n", url)
	// 假設這裡是發送請求，獲取數據
	if strings.Contains(url, "url2") {
		// 假設請求 url2 時出現錯誤
		time.Sleep(time.Second*2)
		return "", errors.New("請求出錯")


	} else if strings.Contains(url, "url3") {
		// 假設 請求 url3 需要1秒
		select {
		case <- errCtx.Done():
			ret, err = "", errors.New("請求3被取消")
			return
		case <- time.After(time.Second*3):
			fmt.Printf("請求 %s 結束....\n", url)
			return "success3", nil
		}
	} else {
		select {
		case <- errCtx.Done():
			ret, err = "", errors.New("請求1被取消")
			return
		case <- time.After(time.Second):
			fmt.Printf("請求 %s 結束....\n", url)
			return "success1", nil
		}
	}

}

執行結果：

請求 http://localhost/url2 開始....
請求 http://localhost/url3 開始....
請求 http://localhost/url1 開始....
請求 http://localhost/url1 結束....
eg.Wait error: 請求出錯

2、errgroup源碼分析

看了上面的例子，我們對errgroup有了一定瞭解，接下來，我們一起看看errgroup做了那些封裝。

2.1 errgroup.Group

errgroup.Group源碼如下：

// A Group is a collection of goroutines working on subtasks that are part of
// the same overall task.
//
// A zero Group is valid, has no limit on the number of active goroutines,
// and does not cancel on error.
type Group struct {
  // context 的 cancel 方法
	cancel func()

	wg sync.WaitGroup
	
  //傳遞信號的通道，這裡主要是用於控制併發創建 goroutine 的數量
  //通過 SetLimit 設置過後，同時創建的goroutine 最大數量為n
	sem chan token
	
  // 保證只接受一次錯誤
	errOnce sync.Once
  // 最先返回的錯誤
	err     error
}

看結構體中的內容，發現比原生的sync.WaitGroup多了下麵的內容：

  • cancel func()
  • sem chan token
  • errOnce sync.Once
  • err error

2.2 WithContext 方法

// WithContext returns a new Group and an associated Context derived from ctx.
//
// The derived Context is canceled the first time a function passed to Go
// returns a non-nil error or the first time Wait returns, whichever occurs
// first.
func WithContext(ctx context.Context) (*Group, context.Context) {
	ctx, cancel := context.WithCancel(ctx)
	return &Group{cancel: cancel}, ctx
}

方法邏輯還是比較簡單的，主要做了兩件事：

  • 使用context的WithCancel()方法創建一個可取消的Context
  • 將context.WithCancel(ctx)創建的 cancel賦值給 Group中的cancel

2.3 Go

1.2 最後一個例子說，不用手動去執行 cancel 的原因就在這裡。

g.cancel() //這裡就是為啥不用手動執行 cancel的原因

// Go calls the given function in a new goroutine.
// It blocks until the new goroutine can be added without the number of
// active goroutines in the group exceeding the configured limit.
//
// The first call to return a non-nil error cancels the group's context, if the
// group was created by calling WithContext. The error will be returned by Wait.
func (g *Group) Go(f func() error) {
	if g.sem != nil {
    //往 sem 通道中發送空結構體，控制併發創建 goroutine 的數量
		g.sem <- token{}
	}

	g.wg.Add(1)
	go func() {
    // done()函數的邏輯就是當 f 執行完後，從 sem 取一條數據，並且 g.wg.Done()
		defer g.done()

		if err := f(); err != nil {
			g.errOnce.Do(func() { // 這裡就是確保 g.err 只被賦值一次
				g.err = err
				if g.cancel != nil {
					g.cancel() //這裡就是為啥不用手動執行 cancel的原因
				}
			})
		}
	}()
}

2.4 TryGo

看註釋，知道此函數的邏輯是：當正在執行的goroutine數量小於通過SetLimit()設置的數量時，可以啟動成功，返回 true，否則啟動失敗，返回false。

// TryGo calls the given function in a new goroutine only if the number of
// active goroutines in the group is currently below the configured limit.
//
// The return value reports whether the goroutine was started.
func (g *Group) TryGo(f func() error) bool {
	if g.sem != nil {
		select {
		case g.sem <- token{}: // 當g.sem的緩衝區滿了過後，就會執行default，也代表著未啟動成功
			// Note: this allows barging iff channels in general allow barging.
		default:
			return false
		}
	}
  
  //----主要看上面的邏輯，下麵的邏輯和Go中的一樣-------

	g.wg.Add(1)
	go func() {
		defer g.done()

		if err := f(); err != nil {
			g.errOnce.Do(func() {
				g.err = err
				if g.cancel != nil {
					g.cancel()
				}
			})
		}
	}()
	return true
}

2.5 Wait

代碼邏輯很簡單，這裡主要註意這裡：

//我看這裡的時候，有點疑惑，為啥這裡會去調用 cancel()方法呢？
//這裡是為了代碼的健壯性，用 context.WithCancel() 創建得到的 cancel，在代碼執行完畢之前取消是一個好習慣
g.cancel()

// Wait blocks until all function calls from the Go method have returned, then
// returns the first non-nil error (if any) from them.
func (g *Group) Wait() error {
  g.wg.Wait() //通過 g.wg.Wait() 阻塞等待所有的 goroutine 執行完
	if g.cancel != nil {
    //我看這裡的時候，有點疑惑，為啥這裡會去調用 cancel()方法呢？
    //這裡是為了代碼的健壯性，用 context.WithCancel() 創建得到的 cancel，在代碼執行完畢之前取消是一個好習慣
 		g.cancel()
	}
	return g.err
}

2.6 SetLimit

看代碼的註釋，我們知道：SetLimit的邏輯主要是限制同時執行的 goroutines 的數量為n，當n小於0時，沒有限制。如果有運行的 goroutine，調用此方法會報錯。

// SetLimit limits the number of active goroutines in this group to at most n.
// A negative value indicates no limit.
//
// Any subsequent call to the Go method will block until it can add an active
// goroutine without exceeding the configured limit.
//
// The limit must not be modified while any goroutines in the group are active.
func (g *Group) SetLimit(n int) {
	if n < 0 {
		g.sem = nil
		return
	}
	if len(g.sem) != 0 {
		panic(fmt.Errorf("errgroup: modify limit while %v goroutines in the group are still active", len(g.sem)))
	}
	g.sem = make(chan token, n)
}

3、errgroup 容易忽視的坑

這個坑是看別人的記錄看到的，對errgroup不太熟悉時，是不小心確實容易掉進去，所以摘抄了過來，如果侵權，請聯繫刪除，謝謝！

原文鏈接：併發編程包之 errgroup

需求:

開啟多個Goroutine去緩存中設置數據，同時開啟一個Goroutine去非同步寫日誌，很快我的代碼就寫出來了：

package main

import (
	"context"
	"errors"
	"fmt"
	"golang.org/x/sync/errgroup"
	"time"
)

func main()  {
	g, ctx := errgroup.WithContext(context.Background())

	// 單獨開一個協程去做其他的事情，不參與waitGroup
	go WriteChangeLog(ctx)

	for i:=0 ; i< 3; i++{
		g.Go(func() error {
			return errors.New("訪問redis失敗\n")
		})
	}
	if err := g.Wait();err != nil{
		fmt.Printf("appear error and err is %s",err.Error())
	}
	time.Sleep(1 * time.Second)
}

func WriteChangeLog(ctx context.Context) error {
	select {
	case <- ctx.Done():
		return nil
	case <- time.After(time.Millisecond * 50):
		fmt.Println("write changelog")
	}
	return nil
}

結果：

appear error and err is 訪問redis失敗

代碼看著沒有問題，但是日誌一直沒有寫入。這是為什麼呢？

其實原因就是因為這個ctx是errgroup.WithContext方法返回的一個帶取消的ctx，我們把這個ctx當作父context傳入WriteChangeLog方法中了，如果errGroup取消了，也會導致上下文的context都取消了，所以WriteChangelog方法就一直執行不到。

這個點是我們在日常開發中想不到的，所以需要註意一下～。

解決方法：

解決方法就是在 go WriteChangeLog(context.Background()) 傳入新的ctx

參考資料：

八. Go併發編程--errGroup

併發編程包之 errgroup

上面這個案例中講了一個容易忽視的坑，大家可以看看

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

-Advertisement-
Play Games
更多相關文章
  • 【FAQ】關於JavaScript版本的華為地圖服務Map的點擊事件與Marker的點擊事件存在衝突的解決方案
    一． 問題描述 創建地圖對象，並添加marker標記，對map和marker均添加了點擊事件； <body> <script> function initMap() { // 創建地圖對象 const map = new HWMapJsSDK.HWMap(document.getElementByI ...
  • Claude：除ChatGPT外的另一種選擇
    前言 Claude 是 Anthropic 開發的人工智慧產品。Anthropic 是由 11 名前 OpenAI 員工於 2022 年創立的人工智慧公司，旨在構建安全、可解釋和有益於人類的人工智慧系統。Claude 是該公司的第一個產品，得到了谷歌 3 億美元的投資。 與 ChatGPT 有什麼區 ...
  • VSCode Webview中如何實現點擊下載圖片
    眾所周知，在一個普通的HTML頁面中，如果要實現一個鏈接點擊後下載圖片，只需要在頁面上放一個<a>標簽，然後將屬性href的值指向圖片的URL或者Base64字元串就可以了。或者按照stackoverflow上提供的方法動態創建<a>標簽來完成圖片的下載動作。不過原理都是相同的。 但是這個方法在VS ...
  • vue-router3.x和vue-router4.x相互影響的問題記錄
    背景 項目中有一個系統使用的微前端，主站使用是vue2實現的，使用的是vue-router3.x。子應用有使用vue3實現的，使用的為vue-router4.x。 該子應用中的頁面A有通過操作按鈕觸發跳轉到其他子應用頁面B的需求，此時使用的是vue-router4.x的編程式導航API。 當通過點擊 ...
  • 記錄-Vue移動端日曆設計與實現
    這裡給大家分享我在網上總結出來的一些知識，希望對大家有所幫助 工作中遇到一個需求是根據日曆查看某一天/某一周/某一月的睡眠報告，但是找了好多日曆組件都不是很符合需求，只好自己手寫一個日曆組件，順便記錄一下。 先看看UI給的設計圖和,需求是有數據的日期做標記,可以查看某一周/某一月的數據,周數據不用自 ...
  • JS中的Map、Set、WeakMap和WeakSet
    在JavaScript中，Map、Set、WeakMap和WeakSet是四個不同的數據結構，它們都有不同的特點和用途： 1. Map ：Map是一種鍵值對的集合，其中的鍵和值可以是任意類型的。與對象類似，它們可以通過鍵來訪問值。不同之處在於，Map可以使用任意類型作為鍵，而對象只能使用字元串或Sy ...
  • 系統認知篇：防腐層、門面模式及適配模式的本質
    門面模式和適配器模式是代碼級的設計模式，而防腐層本質是一種防禦型策略，在更高的層級對系統進行解耦。通常情況下，防腐層包含一系列的門面類和適配器類以及一些轉換器類。 ...
  • 5 種微服務網關介紹
    微服務網關作為微服務後端服務的統一入口，它可以統籌管理後端服務，主要分為數據平面和控制平面： 數據平面主要功能是接入用戶的HTTP請求和微服務被拆分後的聚合。使用微服務網關統一對外暴露後端服務的API和契約，路由和過濾功能正是網關的核心能力模塊。另外，微服務網關可以實現攔截機制和專註跨橫切麵的功能... ...
一周排行
    -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.數據驗證 在伺服器端進行嚴格的數據驗證，確保接收到的數據符合預期格 ...
所有分類