前言

今天也是偏實戰的內容，作為一個并發復習課，很簡單，我們來看看怎樣實現一個并發任務 manager。

在微服務的場景下，我們有很多任務的執行是沒有明確的先后順序的，比如一個接口同時要做到任務 A 和 任務 B，兩個任務分別拿到一些數據，最后組裝裁剪后通過接口下發。

此時，A 和 B 兩個任務沒有依賴關系，如果我們串行來執行，會拖慢整個任務的執行節奏，用并發的方式來優化是一個方向。

那怎么實現呢？

errgroup

一個常見的想法是用 errgroup，我們之前也介紹過?Golang errgroup 設計和原理解析。

今天我們不打算用這種實現，希望用更加基礎的組件來引發思考，看看如何活用 sync 包提供的基礎能力。另外一點是 errgroup 也有他的缺陷，如果在啟動的協程中沒有手動 recover，那么一旦在我們的任務中出現 panic，整個程序就 crash 了。

這一點還是很有爭議的，很多開發者認為這是符合預期的，也有一些開發者希望在 New 一個 errgroup 的時候能夠提供 option 控制是否來 recover。近期還有兩個 issue 在進行激烈的討論，目前看沒有定論。

感興趣的同學可以看下這兩個 issue：

• x/sync/errgroup: why not recover the fn's err in errgroup #40484
• proposal: x/sync/errgroup: propagate panics and Goexits through Wait #53757

需求拆解

ok，我們來試著用 sync 包基礎能力來實現一個簡單的并行任務 manager。首先我們分析下需求。

• 一定要能做到并發執行各個任務，開多個協程，而不是在一個 main goroutine 里串行執行各個任務；
• 并發安全，我們當然不希望出現數據異常，不希望并發執行任務導致最后程序因為 runtime error 而掛掉；
• 如果多個任務都失敗，只返回一個 error 即可；
• 能夠 recover from panic，不需要開發者使用的時候再手動去寫 recover 邏輯；
• 性能有保障。

并發執行這一點我們可以借助 sync.WaitGroup 的能力，每次啟動一個goroutine，WaitGroup 就加 1，在 defer 里完成 Done，啟動所有 goroutine 之后，等著 Wait 返回結果即可。常規的能力復用。

需要額外處理的地方在于，怎么實現多個線程只有一個 error 能賦值，以及 recover 的適配。

實戰代碼

我們理一下思路，看看代碼怎么寫。

Job

首先一定需要定義一個通用的函數簽名，使得開發者能夠傳入自己要執行的并發任務。

type Job interface {
	Do(ctx context.Context, param interface{}) error
	Name() string
}

JobManager

我們的 job manager 現階段可以簡單實現，只是一組 Job 的集合：

type JobManager []Job

錯誤處理

要達到只有一個 error 賦值，且不出現 race condition，有兩個方案：

• sync.Mutex 加鎖；
• sync.Once 只執行一次。

當然，什么時候我們都可以用一把大鎖解決問題，但它的性能不會很好，能用原子操作解決的盡量還是不要用 Mutex，這里參照 errgroup，我們可以用一個 Once 對象來控制只賦值一次。

panic 恢復可以直接在 defer 里面 recover 即可，需要能帶出來 stack trace，把它變成一個 error 賦值

及時退出

有時候我們這個并發任務數量非常多，可能還沒創建完 goroutine，某個先創建的任務就已經掛了，這時候需要有一個全局的信號，終止后續的 goroutine 創建。這一點用原子操作就能實現。

完整代碼

把上面的分析落地，這樣我們就實現了一個帶上了 recover 能力，以及終止能力的的 errgroup。

package main
import (
	"context"
	"errors"
	"fmt"
	"sync"
	"sync/atomic"
)
type Job interface {
	Do(ctx context.Context, param interface{}) error
	Name() string
}
type JobManager []Job

func (mgr JobManager) Execute(ctx context.Context, param interface{}) error {
	var (
		stop    int32 = 0
		err     error
		wg      sync.WaitGroup
		errOnce sync.Once
	)

	for _, job := range mgr {
		if atomic.LoadInt32(&stop) > 0 {
			break
		}

		wg.Add(1)
		go func(j Job) {
			defer func() {
				wg.Done()
				if r := recover(); r != nil {
					errMsg := fmt.Sprintf("JobManager panic: job: %v, reason: %v", j.Name(), r)
					nerr := errors.New(errMsg)
					errOnce.Do(func() {
						if err == nil {
							err = nerr
						}
					})
					atomic.AddInt32(&stop, 1)
				}
			}()
			nerr := j.Do(ctx, param)
			if nerr != nil {
				atomic.AddInt32(&stop, 1)
				errOnce.Do(func() {
					if err == nil {
						err = nerr
					}
				})
			}
		}(job)
	}
	wg.Wait()
	return err
}

使用方法也很簡單：

var mgr = JobManager{
	AJob, BJob, CJob, // 這里的各個 Job 需要實現一開始我們定義的接口
}
err := mgr.Execute(ctx, param)

這里我們需要定義統一的 param interface{}，建議是一個接口，各個 Job 執行完畢后如果有需要寫入的數據，可以調用 param 的 Setter 方法寫入，最后直接拿 param 來做后續邏輯。

小結

今天我們用 sync.Once，以及 sync.WaitGroup 的能力實現了一個簡易的并發任務調度器，希望能夠幫助大家回顧一下此前介紹的并發相關概念和用法。其實并發管理這一點很多時候我們會存在依賴，這時候可能需要將多個 job 分層，或者梳理出來拓撲關系來執行，我們今天只是簡單入門，復習一下相關知識。

建議大家回顧一下此前對于 once 以及 errgroup 的源碼解析，相信你會更能融會貫通。

• Golang errgroup 設計和原理解析
• 解析 Golang sync.Once 用法和原理