本篇内容主要讲解“Golang中怎么手写一个简单的并发任务manager”,感兴趣的朋友不妨来看看。本文介绍的方法操作简单快捷,实用性强。下面就让小编来带大家学习“Golang中怎么手写一个简单的并发任务manager”吧!
需求拆解
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 来做后续逻辑。