Go语言同步与异步执行多个任务封装详解(Runner和RunnerAsync)

概述前言同步适合多个连续执行的，每一步的执行依赖于上一步 *** 作，异步执行则和任务执行顺序无关(如从10个站点抓取数据)

前言

同步适合多个连续执行的，每一步的执行依赖于上一步 *** 作，异步执行则和任务执行顺序无关(如从10个站点抓取数据)

同步执行类RunnerAsync

支持返回超时检测，系统中断检测

错误常量定义

//超时错误var ErrTimeout = errors.New("received timeout")// *** 作系统系统中断错误var ErrInterrupt = errors.New("received interrupt")

实现代码如下

package taskimport ( "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 }} //获取执行完的errorfunc (this *Runner) GetErrs() []error { return this.errs}

使用方法　　　　

Add添加一个任务，任务为接收int类型的一个闭包

Start开始执行伤，返回一个error类型，nil为执行完毕,ErrTimeout代表执行超时，ErrInterrupt代表执行被中断(类似Ctrl + C *** 作)

测试示例代码

package taskimport ( "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(),) 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 taskimport ( "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,lock:  sync.Mutex{},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 }} //获取执行完的errorfunc (this *Runner) GetErrs() []error { return this.errs}

使用方法　　　　

Add添加一个任务，任务为接收int类型，返回类型error的一个闭包

Start开始执行伤，返回一个error类型，nil为执行完毕,ErrTimeout代表执行超时，ErrInterrupt代表执行被中断(类似Ctrl + C *** 作)

getErrs获取所有的任务执行结果

测试示例代码

package taskimport ( "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(),) 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>]

总结

以上就是这篇文章的全部内容了，希望本文的内容对大家的学习或者工作具有一定的参考学习价值，如果有疑问大家可以留言交流，谢谢大家对编程小技巧的支持。

您可能感兴趣的文章:go语言同步教程之条件变量 总结

以上是内存溢出为你收集整理的Go语言同步与异步执行多个任务封装详解(Runner和RunnerAsync)全部内容，希望文章能够帮你解决Go语言同步与异步执行多个任务封装详解(Runner和RunnerAsync)所遇到的程序开发问题。

如果觉得内存溢出网站内容还不错，欢迎将内存溢出网站推荐给程序员好友。