Golang 手写一个简单的并发任务 manager
作者:ag9920??????? 发布时间:2024-02-21 00:15:31
前言
今天也是偏实战的内容,作为一个并发复习课,很简单,我们来看看怎样实现一个并发任务 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 用法和原理
来源:https://juejin.cn/post/7137148885574615077
猜你喜欢
- 【尊重原创,转载请注明出处】https://blog.csdn.net/guyuealian/article/details/7967225
- 共享标签默认情况下,git push 命令并不会传送标签到远程仓库服务器上。在创建完标签后,你必须显式地(手动)推送标签到远程服务
- MySQL语句优化的基本原则:◆1、使用索引来更快地遍历表。缺省情况下建立的索引是非群集索引,但有时它并不是最佳的。在非群集索引下,数据在物
- Python中的模块(.py文件)在创建之初会自动加载一些内建变量,__name__就是其中之一。Python模块中通常会定义很多变量和函数
- 本文实例讲述了PHP实现数组根据某个字段进行水平合并,横向合并。分享给大家供大家参考,具体如下:PHP数组水平合并,横向合并,两条数据合并成
- 本文实例讲述了python实现把二维列表变为一维列表的方法。分享给大家供大家参考,具体如下:c = [[1,2,3], [4,5,6], [
- 从PJBlog 2.7开始,验证码的功能就很好很强大了,但是同时也给手工输入带来了不小的麻烦——经常输错。之前我写了一个《自己写的一个PJB
- 目录一、socketserver实现并发二、验证客户端合法性一、socketserver实现并发tcp协议的socket是只能和一个客户端通
- 国庆假期快到了,想查查还有几天几小时到假期,这对程序员
- 作业备份,不是备份数据库,是备份作业。 我的方法是把作业导出成文件备份起来,因为当你服务器维护的多了的时候很多你的作业 就很成问题,很麻烦。
- Python偏函数Python偏函数和我们之前所学习的函数传参中的缺省参数有些类似,但是在实际应用中还是有所区别的,下面通过模拟一个场景一步
- 阅读目录什么是设计模式单体模式:工厂模式:单例模式观察者模式(发布订阅模式)策略模式模板模式代理模式外观模式设计模式太多了,貌似有23种,其
- create database myIndexDemo go use myIndexDemo go create table ABC ( A
- 要说基本操作,大家基本都会,但是有时候,有些操作使用小技巧会节省很多时间。本篇描述的就是使用dict小技巧,保存到数据库,用来节省大家编码的
- 应用正则表达式的全局匹配,可以匹配出字符出现的次数,比较这些次数,将最大的保存并返回。代码如下: var countMost = funct
- 采用Appium进行自动化的功能性测试最酷的一点是,你可以使用具有最适合你的测试工具的任何一门语言来写你的测试代码。大家选择最多
- 看了OReilly.JavaScript.The.Definitive.Guide.5th.Edition.Aug.2006里的cookie
- 绘制图形利用opencv提供的绘制图形api可以轻松在图像上绘制各种图形,比如直线,矩形,圆,椭圆等图形。line(img,pt
- HTML与CSS在Flash中的应用:不小心看到同事Den在弄个小东西:在Flash里使用HTML和CSS,代码是这样:var m
- 什么是Dynamic HTML 今天我们以问答的形式来讲述什麽是Dynamic Html。问:亲爱的网猴,我经常看到讲述有关“Dynamic