«

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

时间:2024-4-13 17:00     作者:韩俊     分类: Go语言


本篇内容主要讲解“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 来做后续逻辑。

标签: golang

热门推荐