Go 語言并發模式實踐

在 Go 語言并發編程中，合理的并發模式能顯著提升程序的可維護性和性能。本文將深入解析三種典型的并發模式實現，通過具體案例展示如何優雅地管理任務生命周期、資源池和工作 goroutine 池。

一、runner 模式：任務生命周期管理

在定時任務、批處理等場景中，我們需要對任務執行時間進行控制，并在收到中斷信號時安全終止任務。runner 模式通過通道和超時機制實現了這一需求。

1. 核心實現原理

runner 模式的核心在于通過三個通道協同管理任務狀態：

interrupt通道接收操作系統中斷信號
complete通道報告任務完成狀態
timeout通道控制任務執行超時

下面是 runner 包的核心實現：

// Runner 管理任務執行生命周期
type Runner struct {interrupt chan os.Signal    // 接收中斷信號complete  chan error       // 任務完成通知timeout   <-chan time.Time // 超時控制tasks     []func(int)      // 任務列表closed    bool             // 運行狀態
}// New 創建新的Runner實例
func New(d time.Duration) *Runner {return &Runner{interrupt: make(chan os.Signal, 1),complete:  make(chan error),timeout:   time.After(d),}
}// Add 添加任務到Runner
func (r *Runner) Add(tasks ...func(int)) {r.tasks = append(r.tasks, tasks...)
}// Start 啟動任務執行并監視狀態
func (r *Runner) Start() error {// 注冊中斷信號處理signal.Notify(r.interrupt, os.Interrupt)// 啟動任務執行goroutinego func() {r.complete <- r.run()}()// 等待任務完成或超時select {case err := <-r.complete:return errcase <-r.timeout:return errors.New("任務執行超時")}
}// run 按順序執行注冊的任務
func (r *Runner) run() error {for id, task := range r.tasks {// 檢查是否收到中斷信號if r.gotInterrupt() {return errors.New("收到中斷信號")}// 執行任務task(id)}return nil
}// gotInterrupt 檢測中斷信號
func (r *Runner) gotInterrupt() bool {select {case <-r.interrupt:signal.Stop(r.interrupt)return truedefault:return false}
}

2. 應用場景示例

以下是使用 runner 模式實現定時任務的案例，任務將在 3 秒內執行，超時或收到中斷時終止：

func main() {log.Println("開始執行任務...")// 創建3秒超時的Runnerr := runner.New(3 * time.Second)// 添加三個任務r.Add(func(id int) {log.Printf("任務 %d 執行中...", id)time.Sleep(1 * time.Second)},func(id int) {log.Printf("任務 %d 執行中...", id)time.Sleep(2 * time.Second)},func(id int) {log.Printf("任務 %d 執行中...", id)time.Sleep(3 * time.Second)},)// 執行任務并處理結果if err := r.Start(); err != nil {switch err {case errors.New("任務執行超時"):log.Println("任務超時，終止執行")case errors.New("收到中斷信號"):log.Println("收到中斷，終止執行")}}log.Println("任務處理完成")
}

3. 關鍵特性解析

超時控制：通過time.After設置任務整體執行超時時間
中斷處理：利用signal.Notify捕獲系統中斷信號
任務順序執行：按添加順序依次執行任務，適合有依賴關系的場景
優雅退出：無論超時還是中斷，都能確保資源釋放

二、pool 模式：資源池管理

在數據庫連接、文件句柄等資源管理場景中，資源池模式能有效復用資源，避免頻繁創建和銷毀帶來的性能損耗。

1. 資源池核心設計

pool 模式通過有緩沖通道實現資源的獲取與釋放，確保資源復用：

// Pool 管理可復用資源池
type Pool struct {m        sync.Mutex          // 互斥鎖保護資源池resources chan io.Closer     // 資源通道factory  func() (io.Closer, error) // 資源創建工廠closed   bool                // 資源池狀態
}// New 創建新的資源池
func New(fn func() (io.Closer, error), size uint) (*Pool, error) {if size <= 0 {return nil, errors.New("資源池大小不能小于1")}return &Pool{factory:   fn,resources: make(chan io.Closer, size),}, nil
}// Acquire 從資源池獲取資源
func (p *Pool) Acquire() (io.Closer, error) {select {// 有空閑資源時直接獲取case r, ok := <-p.resources:if !ok {return nil, errors.New("資源池已關閉")}return r, nil// 無空閑資源時創建新資源default:return p.factory()}
}// Release 釋放資源回池
func (p *Pool) Release(r io.Closer) {p.m.Lock()defer p.m.Unlock()// 池已關閉時直接關閉資源if p.closed {r.Close()return}// 嘗試將資源放回池，滿時關閉資源select {case p.resources <- r:log.Println("資源放回池")default:log.Println("資源池已滿，關閉資源")r.Close()}
}// Close 關閉資源池并釋放所有資源
func (p *Pool) Close() {p.m.Lock()defer p.m.Unlock()if p.closed {return}p.closed = true// 關閉通道并釋放資源close(p.resources)for r := range p.resources {r.Close()}
}

2. 數據庫連接池應用案例

以下是使用 pool 模式管理數據庫連接的示例，模擬創建和復用數據庫連接：

// dbConnection 模擬數據庫連接
type dbConnection struct {ID int32
}// Close 實現io.Closer接口
func (db *dbConnection) Close() error {log.Printf("關閉連接 %d\n", db.ID)return nil
}var idCounter int32// createConnection 連接創建工廠
func createConnection() (io.Closer, error) {id := atomic.AddInt32(&idCounter, 1)log.Printf("創建新連接 %d\n", id)return &dbConnection{ID: id}, nil
}func main() {// 創建包含2個連接的資源池p, err := pool.New(createConnection, 2)if err != nil {log.Fatal(err)}defer p.Close()var wg sync.WaitGroupwg.Add(5) // 5個任務競爭2個連接// 模擬5個任務獲取連接for i := 0; i < 5; i++ {go func(taskID int) {defer wg.Done()// 獲取連接conn, err := p.Acquire()if err != nil {log.Fatal(err)}defer p.Release(conn)// 模擬數據庫操作log.Printf("任務 %d 使用連接 %d\n", taskID, conn.(*dbConnection).ID)time.Sleep(time.Duration(rand.Intn(1000)) * time.Millisecond)}(i)}wg.Wait()log.Println("所有任務完成")
}

3. 資源池設計要點

接口抽象：通過io.Closer接口實現資源統一管理
動態擴容：無空閑資源時自動創建新資源
安全釋放：通過互斥鎖保證并發安全
優雅關閉：關閉時釋放所有資源，避免泄漏

三、work 模式：goroutine 池實現

在需要控制并發量的場景中，work 模式通過固定數量的 goroutine 池處理任務，避免創建過多 goroutine 導致資源耗盡。

1. 工作池核心實現

work 模式通過無緩沖通道實現任務與工作 goroutine 的同步：

// Worker 定義工作接口
type Worker interface {Task()
}// Pool 工作goroutine池
type Pool struct {work chan Worker  // 任務通道wg   sync.WaitGroup // 等待組
}// New 創建新的工作池
func New(maxGoroutines int) *Pool {p := Pool{work: make(chan Worker),}p.wg.Add(maxGoroutines)for i := 0; i < maxGoroutines; i++ {go func() {// 從通道獲取任務并執行for w := range p.work {w.Task()}p.wg.Done()}()}return &p
}// Run 提交任務到工作池
func (p *Pool) Run(w Worker) {p.work <- w
}// Shutdown 關閉工作池
func (p *Pool) Shutdown() {close(p.work)p.wg.Wait()
}

2. 任務處理應用案例

以下是使用 work 模式處理批量任務的示例，限制同時運行 3 個 goroutine：

// task 實現Worker接口
type task struct {id int
}func (t task) Task() {log.Printf("任務 %d 開始處理\n", t.id)// 模擬任務處理時間time.Sleep(time.Duration(rand.Intn(2000)) * time.Millisecond)log.Printf("任務 %d 處理完成\n", t.id)
}func main() {// 創建包含3個工作goroutine的池p := work.New(3)defer p.Shutdown()var wg sync.WaitGroupwg.Add(10) // 10個任務// 提交10個任務for i := 0; i < 10; i++ {go func(id int) {defer wg.Done()p.Run(task{id: id})}(i)}wg.Wait()log.Println("所有任務處理完畢")
}

3. 工作池特性分析

固定并發量：通過控制 goroutine 數量避免系統負載過高
任務同步：無緩沖通道保證任務與工作 goroutine 一一對應
簡潔易用：通過接口抽象任務邏輯，解耦業務與并發控制
優雅退出：Shutdown 方法確保所有任務完成后退出

四、三種模式的應用場景對比

模式	核心特性	適用場景	典型案例
runner	任務超時控制與中斷處理	定時任務、批處理作業	數據備份、定時報表生成
pool	資源復用與管理	數據庫連接、文件句柄等資源管理	高并發 Web 服務連接池
work	固定并發量任務處理	批量任務處理、限制并發請求	圖片處理、日志分析