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

转载  2018-01-12   作者:雪山飞猪   我要评论

这篇文章主要给大家介绍了关于Go语言同步与异步执行多个任务封装(Runner和RunnerAsync)的相关资料,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友们下面随着小编来一起学习学习吧。

前言

同步适合多个连续执行的,每一步的执行依赖于上一步操作,异步执行则和任务执行顺序无关(如从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>]

总结

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

相关文章

  • golang的HTTP基本认证机制实例详解

    golang的HTTP基本认证机制实例详解

    这篇文章主要介绍了golang的HTTP基本认证机制,结合实例形式较为详细的分析了HTTP请求响应的过程及认证机制实现技巧,需要的朋友可以参考下
    2016-07-07
  • 深入解析Go语言编程中slice切片结构

    深入解析Go语言编程中slice切片结构

    这篇文章主要介绍了Go语言编程中slice切片结构,其中Append方法的用法介绍较为详细,需要的朋友可以参考下
    2015-10-10
  • Go语言实现的树形结构数据比较算法实例

    Go语言实现的树形结构数据比较算法实例

    这篇文章主要介绍了Go语言实现的树形结构数据比较算法,实例分析了树形结构数据比较算法的实现技巧,具有一定参考借鉴价值,需要的朋友可以参考下
    2015-02-02
  • Golang编程实现生成n个从a到b不重复随机数的方法

    Golang编程实现生成n个从a到b不重复随机数的方法

    这篇文章主要介绍了Golang编程实现生成n个从a到b不重复随机数的方法,结合实例形式分析了Go语言字符串操作及随机数生成的相关操作技巧,需要的朋友可以参考下
    2017-01-01
  • 6行代码快速解决golang TCP粘包问题

    6行代码快速解决golang TCP粘包问题

    在用golang开发人工客服系统的时候碰到了粘包问题,那么什么是粘包呢?下面这篇文章主要给大家介绍了关于如何通过6行代码快速解决golang TCP粘包问题的相关资料,文中通过示例代码介绍的非常详细,需要的朋友可以参考借鉴下面随着小编来一起学习学习吧。
    2018-03-03
  • Golang极简入门教程(三):并发支持

    Golang极简入门教程(三):并发支持

    这篇文章主要介绍了Golang极简入门教程(三):并发支持,本文讲解了goroutine线程、channel 操作符等内容,需要的朋友可以参考下
    2014-10-10
  • Go语言中普通函数与方法的区别分析

    Go语言中普通函数与方法的区别分析

    这篇文章主要介绍了Go语言中普通函数与方法的区别,以实例形式对比分析了普通函数与方法使用时的区别与相关技巧,需要的朋友可以参考下
    2015-02-02
  • golang实现通过smtp发送电子邮件的方法

    golang实现通过smtp发送电子邮件的方法

    这篇文章主要介绍了golang实现通过smtp发送电子邮件的方法,实例分析了Go语言基于SMTP协议发送邮件的相关技巧,需要的朋友可以参考下
    2016-07-07
  • 深入分析golang多值返回以及闭包的实现

    深入分析golang多值返回以及闭包的实现

    相对于C/C++,golang有很多新颖的特性,例如goroutine,channel等等,这些特性其实从golang源码是可以理解其实现的原理。今天这篇文章主要来分析下golang多值返回以及闭包的实现,因为这两个实现golang源码中并不存在,我们必须从汇编的角度来窥探二者的实现。
    2016-09-09
  • GO语言求100以内的素数

    GO语言求100以内的素数

    这篇文章主要介绍了GO语言求100以内的素数,主要通过筛选法来实现,涉及GO语言基本的循环与函数调用方法,需要的朋友可以参考下
    2014-12-12

最新评论