前言
同步适合多个连续执行的,每一步的执行依赖于上一步操作,异步执行则和任务执行顺序无关(如从10个站点抓取数据)
同步执行类RunnerAsync
支持返回超时检测,系统中断检测
错误常量定义
//超时错误 var ErrTimeout = errors.New("received timeout") //操作系统系统中断错误 var ErrInterrupt = errors.New("received interrupt")
实现代码如下
package task import ( "os" "time" "os/signal" "sync" ) //异步执行任务 type Runner struct { //操作系统的信号检测 interrupt chan os.Signal //记录执行完成的状态 complete chan error //超时检测 timeout <-chan time.Time //保存所有要执行的任务,顺序执行 tasks []func(id int) error waitGroup sync.WaitGroup lock sync.Mutex errs []error } //new一个Runner对象 func NewRunner(d time.Duration) *Runner { return &Runner{ interrupt: make(chan os.Signal, 1), complete: make(chan error), timeout: time.After(d), waitGroup: sync.WaitGroup{}, lock: sync.Mutex{}, } } //添加一个任务 func (this *Runner) Add(tasks ...func(id int) error) { this.tasks = append(this.tasks, tasks...) } //启动Runner,监听错误信息 func (this *Runner) Start() error { //接收操作系统信号 signal.Notify(this.interrupt, os.Interrupt) //并发执行任务 go func() { this.complete <- this.Run() }() select { //返回执行结果 case err := <-this.complete: return err //超时返回 case <-this.timeout: return ErrTimeout } } //异步执行所有的任务 func (this *Runner) Run() error { for id, task := range this.tasks { if this.gotInterrupt() { return ErrInterrupt } this.waitGroup.Add(1) go func(id int) { this.lock.Lock() //执行任务 err := task(id) //加锁保存到结果集中 this.errs = append(this.errs, err) this.lock.Unlock() this.waitGroup.Done() }(id) } this.waitGroup.Wait() return nil } //判断是否接收到操作系统中断信号 func (this *Runner) gotInterrupt() bool { select { case <-this.interrupt: //停止接收别的信号 signal.Stop(this.interrupt) return true //正常执行 default: return false } } //获取执行完的error func (this *Runner) GetErrs() []error { return this.errs }
使用方法
Add添加一个任务,任务为接收int类型的一个闭包
Start开始执行伤,返回一个error类型,nil为执行完毕, ErrTimeout代表执行超时,ErrInterrupt代表执行被中断(类似Ctrl + C操作)
测试示例代码
package task import ( "testing" "time" "fmt" "os" "runtime" ) func TestRunnerAsync_Start(t *testing.T) { //开启多核 runtime.GOMAXPROCS(runtime.NumCPU()) //创建runner对象,设置超时时间 runner := NewRunnerAsync(8 * time.Second) //添加运行的任务 runner.Add( createTaskAsync(), createTaskAsync(), createTaskAsync(), createTaskAsync(), createTaskAsync(), createTaskAsync(), createTaskAsync(), createTaskAsync(), createTaskAsync(), createTaskAsync(), createTaskAsync(), createTaskAsync(), createTaskAsync(), ) fmt.Println("同步执行任务") //开始执行任务 if err := runner.Start(); err != nil { switch err { case ErrTimeout: fmt.Println("执行超时") os.Exit(1) case ErrInterrupt: fmt.Println("任务被中断") os.Exit(2) } } t.Log("执行结束") } //创建要执行的任务 func createTaskAsync() func(id int) { return func(id int) { fmt.Printf("正在执行%v个任务\n", id) //模拟任务执行,sleep两秒 //time.Sleep(1 * time.Second) } }
执行结果
同步执行任务 正在执行0个任务 正在执行1个任务 正在执行2个任务 正在执行3个任务 正在执行4个任务 正在执行5个任务 正在执行6个任务 正在执行7个任务 正在执行8个任务 正在执行9个任务 正在执行10个任务 正在执行11个任务 正在执行12个任务 runnerAsync_test.go:49: 执行结束
异步执行类Runner
支持返回超时检测,系统中断检测
实现代码如下
package task import ( "os" "time" "os/signal" "sync" ) //异步执行任务 type Runner struct { //操作系统的信号检测 interrupt chan os.Signal //记录执行完成的状态 complete chan error //超时检测 timeout <-chan time.Time //保存所有要执行的任务,顺序执行 tasks []func(id int) error waitGroup sync.WaitGroup lock sync.Mutex errs []error } //new一个Runner对象 func NewRunner(d time.Duration) *Runner { return &Runner{ interrupt: make(chan os.Signal, 1), complete: make(chan error), timeout: time.After(d), waitGroup: sync.WaitGroup{}, lock: sync.Mutex{}, } } //添加一个任务 func (this *Runner) Add(tasks ...func(id int) error) { this.tasks = append(this.tasks, tasks...) } //启动Runner,监听错误信息 func (this *Runner) Start() error { //接收操作系统信号 signal.Notify(this.interrupt, os.Interrupt) //并发执行任务 go func() { this.complete <- this.Run() }() select { //返回执行结果 case err := <-this.complete: return err //超时返回 case <-this.timeout: return ErrTimeout } } //异步执行所有的任务 func (this *Runner) Run() error { for id, task := range this.tasks { if this.gotInterrupt() { return ErrInterrupt } this.waitGroup.Add(1) go func(id int) { this.lock.Lock() //执行任务 err := task(id) //加锁保存到结果集中 this.errs = append(this.errs, err) this.lock.Unlock() this.waitGroup.Done() }(id) } this.waitGroup.Wait() return nil } //判断是否接收到操作系统中断信号 func (this *Runner) gotInterrupt() bool { select { case <-this.interrupt: //停止接收别的信号 signal.Stop(this.interrupt) return true //正常执行 default: return false } } //获取执行完的error func (this *Runner) GetErrs() []error { return this.errs }
使用方法
Add添加一个任务,任务为接收int类型,返回类型error的一个闭包
Start开始执行伤,返回一个error类型,nil为执行完毕, ErrTimeout代表执行超时,ErrInterrupt代表执行被中断(类似Ctrl + C操作)
getErrs获取所有的任务执行结果
测试示例代码
package task import ( "testing" "time" "fmt" "os" "runtime" ) func TestRunner_Start(t *testing.T) { //开启多核心 runtime.GOMAXPROCS(runtime.NumCPU()) //创建runner对象,设置超时时间 runner := NewRunner(18 * time.Second) //添加运行的任务 runner.Add( createTask(), createTask(), createTask(), createTask(), createTask(), createTask(), createTask(), createTask(), createTask(), createTask(), createTask(), createTask(), createTask(), createTask(), ) fmt.Println("异步执行任务") //开始执行任务 if err := runner.Start(); err != nil { switch err { case ErrTimeout: fmt.Println("执行超时") os.Exit(1) case ErrInterrupt: fmt.Println("任务被中断") os.Exit(2) } } t.Log("执行结束") t.Log(runner.GetErrs()) } //创建要执行的任务 func createTask() func(id int) error { return func(id int) error { fmt.Printf("正在执行%v个任务\n", id) //模拟任务执行,sleep //time.Sleep(1 * time.Second) return nil } }
执行结果
异步执行任务 正在执行2个任务 正在执行1个任务 正在执行4个任务 正在执行3个任务 正在执行6个任务 正在执行5个任务 正在执行9个任务 正在执行7个任务 正在执行10个任务 正在执行13个任务 正在执行8个任务 正在执行11个任务 正在执行12个任务 正在执行0个任务 runner_test.go:49: 执行结束 runner_test.go:51: [<nil> <nil> <nil> <nil> <nil> <nil> <nil> <nil> <nil> <nil> <nil> <nil> <nil> <nil>]
总结
以上就是这篇文章的全部内容了,希望本文的内容对大家的学习或者工作具有一定的参考学习价值,如果有疑问大家可以留言交流,谢谢大家对的支持。
免责声明:本站文章均来自网站采集或用户投稿,网站不提供任何软件下载或自行开发的软件!
如有用户或公司发现本站内容信息存在侵权行为,请邮件告知! 858582#qq.com
白云城资源网 Copyright www.dyhadc.com
暂无“Go语言同步与异步执行多个任务封装详解(Runner和RunnerAsync)”评论...
P70系列延期,华为新旗舰将在下月发布
3月20日消息,近期博主@数码闲聊站 透露,原定三月份发布的华为新旗舰P70系列延期发布,预计4月份上市。
而博主@定焦数码 爆料,华为的P70系列在定位上已经超过了Mate60,成为了重要的旗舰系列之一。它肩负着重返影像领域顶尖的使命。那么这次P70会带来哪些令人惊艳的创新呢?
根据目前爆料的消息来看,华为P70系列将推出三个版本,其中P70和P70 Pro采用了三角形的摄像头模组设计,而P70 Art则采用了与上一代P60 Art相似的不规则形状设计。这样的外观是否好看见仁见智,但辨识度绝对拉满。
更新日志
2024年05月16日
2024年05月16日
- 谭小环.1997-自主【BMG】【WAV+CUE】
- 张震岳《精选 滚石SACD系列》首批限量编号 [SACD ISO][1.1G]
- 童丽《独家爱唱7》头版限量编号24K金碟[DFF][1.6G]
- 韩宝仪《我心若玫瑰》[WAV+CUE][441M]
- 群星《经典老歌百万畅销LP黑胶》10CD[WAV+CUE][6.3G]
- 蔡琴《伤心小站SACD》[DSF][1.1G]
- 群星《青春重置计划 4 珍稀》[320K/MP3][207.12MB]
- 黑帝斯哈迪斯2女巫之杖法杖图鉴及附魔效果一览 法杖图文攻略
- 鸣潮炽霞声骸怎么选择 炽霞声骸搭配及词条选择攻略
- 黑帝斯哈迪斯2姊妹双刃图鉴及附魔效果一览 黑帝斯哈迪斯2全武器图文介绍
- 刘文正.2013-金典歌展2CD【南方】【WAV+CUE】
- 刘珺儿.2001-标准心上人【P.TMUSIC】【WAV+CUE】
- WBG入选沙特电竞世界杯!角逐多个电竞项目
- 《小谢尔顿》最终季大结局首曝片段 成年谢米回归!
- 鸣潮凌阳声骸怎么选择 凌阳声骸搭配及词条选择攻略