golang中channel+error來做非同步錯誤處理有多香

2023-01-20 14:01:00

場景

如下，一次任務起多個協程非同步處理任務，比如同時做服務/redis/mysql/kafka初始化，當某一個協程出現錯誤(初始化失敗)時，程式是停止還是繼續呢？如何記錄錯誤？如何控制優雅的退出全部工作協程呢？

為了解決類似的問題，常見如下三種解決方案：

1.中斷退出並記錄紀錄檔

如下，最簡單粗暴的方式就是，一旦協程中發生錯誤，記錄紀錄檔立即退出，外層如果想取到錯誤可以通過共用全域性變數，單個協程無法控制所有協程動作。

// 出錯中斷協程，列印紀錄檔，退出

func TestSimpleExit(t *testing.T) {
	wg := sync.WaitGroup{}

	wg.Add(1)
	go func() {
		defer wg.Done()

		//do something
		if err := doSomething(); err != nil {
			t.Logf("Error when call doSomething:%vn", err)
			return
		}
	}()

	wg.Wait()
}

2.監控error，可選記錄紀錄檔或退出

前面講過，既然error是值，那就可以在channel中傳輸，可以單獨開一個channel，所有協程錯誤都傳送到這個通道。

	// 資料處理流程
	dataFunc := func(ctx context.Context, dataChan chan int, errChan chan error) {
		defer wg.Done()

		for {
			select {
			case v, ok := <-dataChan:
				if !ok {
					log.Println("Receive data channel close msg!")
					return
				}

				if err := doSomething2(v); err != nil {
					errChan <- err
					continue
				}

				// do ...

			case <-ctx.Done():
				log.Println("Receive exit msg!")
				return
			}
		}
	}

	wg.Add(1)
	go dataFunc(ctx, dataChan, errChan)

	wg.Add(1)
	go dataFunc(ctx, dataChan, errChan)

監控錯誤error通道，統一記錄和退出，一旦檢測到錯誤可以通過ctx通知所有協程退出，這裡可以靈活控制監控到錯誤時的錯誤處理策略(是否記錄紀錄檔/是否退出等)，error通道可以同步或非同步處理。
整體流程如下

非同步監控error

	// 錯誤處理流程，error處理通道非同步等待
	wg.Add(1)
	go func(errChan chan error) {
		defer wg.Done()

		for {
			select {
			case v, ok := <-errChan:
				if !ok {
					log.Println("Receice err channel close msg!")
					return
				}

				// 收到錯誤時，可選擇記錄紀錄檔或退出
				if v != nil {
					t.Logf("Error when call doSomething:%vn", v)
					cancel() // 通知全部退出
					return
				}

			case <-ctx.Done():
				log.Println("Receive exit msg!")
				return
			}
		}

	}(errChan)

	dataChan <- 1
	wg.Wait()

同步監控error

	// 錯誤處理流程，error處理通道同步等待
	for {
		select {
		case v, ok := <-errChan:
			if !ok {
				log.Println("Receice err channel close msg!")
				goto EXIT
			}

			// 收到錯誤時，可選擇記錄紀錄檔或退出
			if v != nil {
				t.Logf("Error when call doSomething:%vn", v)
				cancel()
				goto EXIT
			}

		case <-ctx.Done():
			log.Println("Receive exit msg!")
			goto EXIT
		}
	}

EXIT:
	wg.Wait()

3.官方庫errgroup

考慮到error輸出到通道後統一處理是golang常用手段，官方也針對封裝了一個error處理包，errgroup顧名思義，多個協程的error被當做一個組，一旦某個協程出錯所有協程都退出，只輸出第一個error。

func TestSimpleChannel5(t *testing.T) {
	eg, ctx := errgroup.WithContext(context.Background())

	dataChan := make(chan int)
	defer close(dataChan)

	// 資料處理流程
	dataFunc := func() error {
		for {
			select {
			case v, ok := <-dataChan:
				if !ok {
					log.Println("Receive data channel close msg!")
					return nil
				}

				if err := doSomething2(v); err != nil {
					return err
				}

				// do ...

			// 增加ctx通知完成
			case <-ctx.Done():
				log.Println("Receive exit msg!")
				return nil
			}
		}
	}

	eg.Go(dataFunc)
	eg.Go(dataFunc)
	eg.Go(dataFunc)

	dataChan <- 1

	// 錯誤處理流程，任何一個協程出現error，則會呼叫ctx對應cancel函數，所有相關協程都會退出
	if err := eg.Wait(); err != nil {
		fmt.Printf("Something is wrong->%vn", err)
	}
}

類似上一小節，可以看到errgroup就是結合waitgroup cancel和channel通道封裝的。

4.監控error，全部紀錄檔合併後輸出

同樣是上述場景，有時候我們的需求是返回所有的錯誤(不是第一個錯誤)。

比如在服務啟動時，對 redis、kafka、mysql 等各種資源初始化場景，可以把所有相關資源初始化的錯誤都返回，展示給使用者統一排查。
另一種場景就是在 web 請求中，校驗請求引數時，返回所有引數的校驗錯誤給使用者端的場景。

這種需求，一般考慮使用多錯誤管理(hashicorp/go-multierror庫)，如下一個簡答同步任務演示多錯誤管理，所有返回的錯誤可以通過Append歸併成一個錯誤，實際上是通過error wrap的方式合併起來的，因此也可以使用Is/As判斷巢狀error。

// 多路協程error合併，用於多路check場景
func TestSimpleChannel3(t *testing.T) {

	// 同步執行多個任務，返回error合併
	var err = func() error {
		var result error

		if err := doSomething(); err != nil {
			result = multierror.Append(result, err)
		}

		if err := doSomething2(nil); err != nil {
			result = multierror.Append(result, err)
		}

		return result
	}()

	// 列印輸出
	if err != nil {
		fmt.Printf("%vn", err)
	}

	// 獲取錯誤列表
	if err != nil {
		if merr, ok := err.(*multierror.Error); ok {
			fmt.Printf("%vn", merr.Errors)
		}
	}

	// 判斷是否為某種型別
	if err != nil && errors.Is(err, Error1) {
		fmt.Println("Errors contain error 1")
	}

	// 判斷是否其中一個error能夠轉換成指定error
	var e MyError
	if err != nil && errors.As(err, &e) {
		fmt.Println("One Error can be convert to nyerror")
	}
}

那麼，在起多個非同步任務時，就可以如下處理，返回的多個error通過channel消費合併展示。

func TestSimpleChannel4(t *testing.T) {
	wg := sync.WaitGroup{}

	taskNum := 10
	errChan := make(chan error, taskNum)

	// 非同步執行多個任務
	step := func(stepNum int, errChan chan error) {
		defer wg.Done()
		errChan <- fmt.Errorf("step %d error", stepNum)
	}

	for i := 0; i < taskNum; i++ {
		wg.Add(1)
		go step(i, errChan)
	}

	// 等待任務完成
	go func() {
		wg.Wait()
		close(errChan)
	}()

	// err通道阻塞等待，可能的所有錯誤合併
	var result *multierror.Error
	for err := range errChan {
		result = multierror.Append(result, err)
	}

	// 出現一個錯誤時，選擇記錄紀錄檔或退出
	if len(result.Errors) != 0 {
		log.Println(result.Errors)
	}
}