Golang 手写一个简单的并发任务 Manager

背景与目标

在微服务架构中，经常遇到接口需要同时调用多个独立子任务（如任务 A 和任务 B）的场景。串行执行会显著拖慢响应速度，因此并发优化是必然选择。虽然官方推荐的 errgroup 能满足基本需求，但它在 panic 处理上较为激进：若协程内未手动 recover，任何 panic 都会导致整个程序崩溃。此外，errgroup 缺乏简单的 panic 捕获选项，社区对此争议较大。本文旨在提供一个轻量级替代方案，保留基础能力并增强稳定性。

需求拆解

要实现一个健壮的基础版并发管理器，需满足以下核心要求：

• 并发执行：启动多个 Goroutine 并行运行任务，而非在主协程中串行处理。
• 并发安全：确保多个协程写入状态时不出现数据竞争（Race Condition），避免运行时错误。
• 错误聚合：若多个任务失败，仅返回第一个遇到的错误即可，无需罗列所有错误。
• Panic 恢复：内部自动处理 panic，将其转换为 error 返回，降低使用者的维护成本。
• 性能保障：避免使用重量级锁（Mutex），尽量采用原子操作和轻量级同步原语。

关键设计思路

1. Job 接口定义

首先定义通用的任务接口，允许开发者传入自定义逻辑：

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

2. 错误处理的单写模式

为避免多个协程同时赋值引发竞态条件，有两种方案：加锁或使用 sync.Once。sync.Once 性能更优，可确保错误值仅被赋值一次，符合“只返回一个错误”的需求。

3. Panic 捕获与堆栈信息

在每个协程中使用 defer 配合 recover。捕获到 panic 后，通过 errors.New 包装 panic 信息和 Job 名称，并通过 sync.Once 提交给全局错误变量。

4. 快速终止机制

引入 atomic.AddInt32 作为全局停止信号。当第一个任务失败或发生 panic 时，设置停止标志。后续尚未启动的协程或正在执行的协程（可通过检查上下文或提前判断）可据此终止，减少无效开销。

实战代码实现

完整 Manager 实现

结合上述思路，以下是完整的并发任务管理器代码。它结合了 sync.WaitGroup 等待所有任务完成，以及原子操作和 Once 来控制状态。

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()
				// 尝试从 Panic 中恢复
				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
}

使用方法

定义实现 Job 接口的具体任务，并将它们加入 JobManager 进行执行。建议使用共享的 param 结构体，让各个 Job 完成后通过 Setter 方法更新数据，最后统一处理结果。

var mgr = JobManager{
	AJob, BJob, CJob, // 各 Job 需实现 Job 接口
}

err := mgr.Execute(ctx, param)

小结

通过 sync.Once、sync.WaitGroup 和原子操作，我们构建了一个具备 Panic 恢复能力和快速失败机制的简易并发管理器。这有助于深入理解 Go 并发原语的组合用法。对于复杂的任务依赖拓扑，建议进一步梳理分层逻辑。如需更标准化的行为，仍可参考官方 errgroup 源码，但此方案在无 recover 保护的复杂线上环境中提供了更高的容错性。

以上就是 Golang 手写一个简单并发任务 manager 的详细内容，更多关于 Golang 并发编程的资料请关注本站其它相关文章！