快捷導(dǎo)航

golang基于errgroup實(shí)現(xiàn)并發(fā)調(diào)用的方法

更新時(shí)間：2022年09月05日 08:57:19 作者：夏日白云

這篇文章主要介紹了golang基于errgroup實(shí)現(xiàn)并發(fā)調(diào)用,本文通過實(shí)例代碼給大家介紹的非常詳細(xì)，對(duì)大家的學(xué)習(xí)或工作具有一定的參考借鑒價(jià)值，需要的朋友可以參考下

串行調(diào)用

在用go編寫web/rpc服務(wù)器的時(shí)候，經(jīng)常會(huì)出現(xiàn)需要對(duì)下游多個(gè)/組服務(wù)調(diào)用rpc(或者其他比較耗時(shí)的操作)的情況。
按照自然的寫法，比如對(duì)下游有ABC三個(gè)調(diào)用，串行順著寫，就總共要花費(fèi)TimeA+TimeB+TimeC的時(shí)間：

func Handler(ctx context.Context) {
    var a, b, c respType
    a = A(ctx)
    b = B(ctx)
    c = C(ctx)
}

基于sync.WaitGroup實(shí)現(xiàn)簡(jiǎn)單的并發(fā)調(diào)用

但經(jīng)常地，幾個(gè)rpc相互之間沒有依賴關(guān)系的情況，這時(shí)，我們稍加思考就會(huì)想到使用并發(fā)的方式，同時(shí)發(fā)出請(qǐng)求，阻塞等到所有請(qǐng)求返回，這樣，總體耗時(shí)就變成了Max(TimeA, TimeB, TimeC)，我們可以通過常用的sync.WaitGroup輕松實(shí)現(xiàn)這事：

func Handler(ctx context.Context) {
    var a, b, c respType
   	wg := sync.WaitGroup{}
	wg.Add(3)
	go func() {
		defer wg.Done()
		a = A(ctx)
	}()
	go func() {
		defer wg.Done()
		b = B(ctx)
	}()
	go func() {
		defer wg.Done()
		c = C(ctx)
	}()
	wg.Wait()
}

但是現(xiàn)實(shí)事件是不完美的，尤其是在加入了網(wǎng)絡(luò)這一因素后，我們經(jīng)常會(huì)需要處理調(diào)用失敗的情況，很多情況下，并發(fā)的幾個(gè)操作只要任一失敗，整個(gè)處理就算失敗了，但是由于WaitGroup要等所有調(diào)用都done才能返回，因此調(diào)用時(shí)間是由耗時(shí)最長(zhǎng)的那個(gè)(不一定是失敗的)決定的，如果不是失敗的那個(gè)，其實(shí)就產(chǎn)生了資源浪費(fèi)，如下圖，B最先失敗了，此時(shí)邏輯上已經(jīng)可以返回，但是實(shí)際卻等到了最長(zhǎng)的調(diào)用-A返回了整個(gè)函數(shù)才返回：

func Handler(ctx context.Context) {
    var a, b, c respType
    var errA, errB, errC error
   	wg := sync.WaitGroup{}
	wg.Add(3)
	go func() {
		defer wg.Done()
		a, errA = A(ctx)
	}()
	go func() {
		defer wg.Done()
		b, errB = B(ctx)
	}()
	go func() {
		defer wg.Done()
		c, errC = C(ctx)
	}()
	wg.Wait()
	if errA != nil {
	// ...
	}
	if errB != nil {
	// ...
	}
	if errC != nil {
	// ...
	}
}

基于errgroup.Group實(shí)現(xiàn)并發(fā)調(diào)用

這對(duì)于追求極致的我們來說顯然是不能接受的，我們希望達(dá)到，如果有任意一個(gè)調(diào)用報(bào)錯(cuò)，立刻讓所有調(diào)用返回的效果：

好在，我們有現(xiàn)成的工具可以用，通過引入"golang.org/x/sync/errgroup"，可以輕松實(shí)現(xiàn)上面的目的。

為了使用errgroup，先使用WithContext方法創(chuàng)建一個(gè)Group

wg, groupCtx := errgroup.WithContext(ctx)

返回的第一個(gè)參數(shù)是*errgroup.Group，第二個(gè)則是在子調(diào)用中應(yīng)該使用的context。

然后，使用Go方法調(diào)用所有的并發(fā)方法

	wg.Go(func() error {
		var err error
		a, err = A(groupCtx)
		return err
	})

最后，使用Wait方法等待并發(fā)結(jié)束，返回值是所有子調(diào)用中第一個(gè)非nil的error，全成功的話就是nil。

if err := wg.Wait(); err != nil {
// ...
}

因此整體，我們的代碼差不多就長(zhǎng)這個(gè)樣子

func handler(ctx context.Context) {
    var a, b, c respType
    wg, groupCtx := errgroup.WithContext(ctx)
	wg.Go(func() error {
		var err error
		a, err = A(groupCtx)
		return err
	})
	wg.Go(func() error {
		var err error
		b, err = B(groupCtx)
		return err
	})
	wg.Go(func() error {
		var err error
		c, err = C(groupCtx)
		return err
	})
	if err := wg.Wait(); err != nil {
    // ... 錯(cuò)誤處理
    }
    // 全部成功
}

errgroup內(nèi)部通過封裝了waitGroup和sync.Once實(shí)現(xiàn)了這個(gè)語法糖。

使用時(shí)特別要注意的是，errgroup的提前取消調(diào)用rpc是通過cancel那個(gè)返回的context(即上面的groupCtx)實(shí)現(xiàn)的，因此在所有子調(diào)用中都要實(shí)現(xiàn)監(jiān)聽groupCtx的Done事件。而在正常的rpc框架中都已經(jīng)幫我們實(shí)現(xiàn)了這件事，因此我們只要保證傳進(jìn)去的是groupCtx即可。