串行調(diào)用

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

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

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

但經(jīng)常地，幾個rpc相互之間沒有依賴關(guān)系的情況，這時，我們稍加思考就會想到使用并發(fā)的方式，同時發(fā)出請求，阻塞等到所有請求返回，這樣，總體耗時就變成了Max(TimeA, TimeB, TimeC)，我們可以通過常用的sync.WaitGroup輕松實現(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)實事件是不完美的，尤其是在加入了網(wǎng)絡(luò)這一因素后，我們經(jīng)常會需要處理調(diào)用失敗的情況，很多情況下，并發(fā)的幾個操作只要任一失敗，整個處理就算失敗了，但是由于WaitGroup要等所有調(diào)用都done才能返回，因此調(diào)用時間是由耗時最長的那個(不一定是失敗的)決定的，如果不是失敗的那個，其實就產(chǎn)生了資源浪費，如下圖，B最先失敗了，此時邏輯上已經(jīng)可以返回，但是實際卻等到了最長的調(diào)用-A返回了整個函數(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實現(xiàn)并發(fā)調(diào)用

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

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

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

wg, groupCtx := errgroup.WithContext(ctx)

返回的第一個參數(shù)是*errgroup.Group，第二個則是在子調(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)用中第一個非nil的error，全成功的話就是nil。

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

因此整體，我們的代碼差不多就長這個樣子

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 {
    // ... 錯誤處理
    }
    // 全部成功
}

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

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

總結(jié)

errgroup幫我們封裝了并發(fā)調(diào)用下游時快速失敗的邏輯，我們能很方便地使用它進行業(yè)務(wù)代碼的編寫。使用的關(guān)鍵是一定要記得在子調(diào)用中傳遞WithContext中返回的Context。

好用的工具千千萬，讓我們一個個來掌握！

最新評論