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

来到golang学习网的大家，相信都是编程学习爱好者，希望在这里学习Golang相关编程知识。下面本篇文章就来带大家聊聊《Golang 手写一个简单的并发任务 manager》，介绍一下并发、任务、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 用法和原理

今天关于《Golang 手写一个简单的并发任务 manager》的内容就介绍到这里了，是不是学起来一目了然！想要了解更多关于golang的内容请关注golang学习网公众号！