安科网

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

哈嘿Blog

哈嘿Blog

2018-01-12

关注 关注

前言

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

总结

go语言 tasks

哈嘿Blog

哈嘿Blog

0 关注 0 粉丝 0 动态

关注 关注

相关推荐

详解Go语言中用 os/exec 执行命令的五种方法

在 Golang 中用于执行命令的库是 os/exec，exec.Command 函数返回一个 Cmd 对象，根据不同的需求，可以将命令的执行分为三种情况。直接调用 Cmd 对象的 Run 函数，返回的只有成功和失败，获取不到任何输出的结果。Combine

小科的奋斗 2020-11-16

关于go语言编码需要放到src 文件夹下的问题

配置GOPATH的用意是为了方便项目的部署和构建，以及可以直接使用go get 命令下载第三方的包到自己的项目的src下和相关的执行文件bin目录，和中间文件pkg. 问1：index entire GOPATH:如果你选中那么我就把你在环境变量中配置的G

wfs 2020-10-29

GO语言gin框架实现管理员认证登陆接口

后台用户登录验证功能是很多项目的必须要有的逻辑 , 也是常见的技术需求 .

hellosunshine 2020-10-27

Go语言实现猜数字小游戏

利用Go语言实现一个四位数字，而且不重复的猜数字游戏，解题思路是先利用rand随机函数产生一个四位数，然后通过切片进行引用传递，设定用户可以输入的数字，最后分别遍历切片四位数字与产生的四位数是否相同。break //跳出最外层循环。randSlice:=m

哈嘿Blog 2020-10-26

一篇文章带你了解Go语言基础之指针

作为21世纪的C语言，Go可能多方面思考，最终还是保留了指针。因为Go语言是类C语言，所以为了提高效率，仍然保留了指针。但是如果没有接触过指针，可能学起来指针需要多多思考一下了。指针通常情况下，也被称为小白劝退神奇。栈内存在存储上，只能存一些简单的东西，比

bearhoopIT之道 2020-11-11

属于新十年的开发语言：Go语言可能很快会取代Python

30年前，Python首次亮相，它花了20年才得到开发者的赞赏。如今，它已经成为开发者第二喜爱的语言。尤其是在过去五年里，Python用户激增，成为机器学习和数据科学开发者的首选语言。在未来几年里，Python在这些领域的主导地位必然是很难被动摇的。但与较

文山羊 2020-11-07

一篇文章带你了解Go语言基础之函数（上篇）

Hey，大家好呀，我是星期八，这次咱们继续学习Go基础之函数叭。函数，在所有编程语言中都叫函数，Java，PHP，Python，JS等，统一都叫函数。一般是这样形容的:函数可以将重复的，或者特定功能的，封装成一个方便调用的东西。函数命名尽量以驼峰命名，例如

jaminliu0 2020-10-25

Go语言在极小硬件上的运用（二）

在本文的第一部分的结尾，我承诺要写关于接口的内容。我不想在这里写有关接口或完整或简短的讲义。相反，我将展示一个简单的示例，来说明如何定义和使用接口，以及如何利用无处不在的io.Writer接口。还有一些关于反射reflection和半主机semihosti

railssky 2020-10-25

Go语言在极小硬件上的运用（一）

Go 语言，能在多低下的配置上运行并发挥作用呢？我购买它的理由有三个。首先，我从未接触过 STM320 系列的开发板。其次，STM32F10x 系列使用也有点少了。STM320 系列的 MCU 很便宜，有更新一些的外设，对系列产品进行了改进，问题修复也

小科的奋斗 2020-10-24

一篇文章带你了解Go语言基础之字节类型

Hey，大家好呀，我是星期八，这次咱们继续学习Go基础之字节类型吧。可以看到，在Go中，字符串其实就是指向了一个连续的内存地址，并且记录了长度，读取是一次性读取出来的。字符串是一个个字符拼接成的，反过来，也是一个个字符拼接成的字符串，而一个个的字符，就被称

HELLOBOY 2020-10-23

一篇文章带你了解Go语言基础之变量

Hey，大家好呀，我是星期八，这次咱们继续学习Go基础之变量叭。简单点说，我们写的程序默认数据都是保存在内存条中的，我们不可能直接通过地址找到这个变量，因为地址太长了，而且不容易记。通常我们是通过变量名找到对应的地址上面的值，再进行一系列操作。现在不管是什

牧 2020-10-22

一篇文章带你了解Go语言基础之数组

Hey，大家好呀，我是星期八，这次咱们继续学习Go基础之数组扒。//fmt.Println//结果:[0000000000]声明不赋值，int类型默认是0，其他类型也有默认值。//varname_list=[...]int{1，2，4，5，19}//...

狼窝 2020-10-22

专为人工智能和数据科学而生的Go语言，或将取代Python

时间回到30年前，Python 首次亮相。花了20年时间，Python才获得开发者的认可。来到 2019 年，Python已经成为开发人员第二喜欢的语言。在过去的5年时间里，Python 成为了机器学习和数据科学开发人员的首选编程语言。可以预见的是，在未

ciqingloveless 2020-10-15

Go语言酷的一些东西

Go是一种相对较新的语言，由Google的Robert Griesemer，Rob Pike和Ken Thompson于2009年之前创建。它是开源的，因此任何人都可以为此做出贡献并提出新功能。替代C ++似乎主要是为了使Google的软件工程师的工作更

专注前端开发 2020-10-12

Go语言使用swagger生成接口文档的方法

Swagger本质上是一种用于描述使用JSON表示的RESTful API的接口描述语言。Swagger与一组开源软件工具一起使用，以设计、构建、记录和使用RESTful Web服务。Swagger包括自动文档，代码生成和测试用例生成。可是编写接口文档历来

哈嘿Blog 2020-09-08

Go语言中如何通过方法为类型添加行为

数十年以来， 传统的面向对象语言总是说方法属于类， 但 Go 不是这样做的: 它提供了方法， 但是并没有提供类和对象。乍一看， 这种做法似乎有些奇怪， 甚至可以说有点儿疯狂， 但实际上 Go 的方法比以往其他语言的方法都要灵活。我们可以将方法与同一个包中声

linmufeng 2020-09-08

Go语言文件读取的一些总结

最常见的比如直接对文件本身进行Read和Write； 除此之外，还可以使用bufio库的流式处理以及分片式处理；如果文件较小，使用ioutil也不失为一种方法。面对这么多的文件处理的方式，那么初学者可能就会有困惑：我到底该用那种？它们之间有什么区别？笔者试

qidiantianxia 2020-09-15

详解Go语言中关于包导入必学的 8 个知识点

在 Go 语言中，一个包可包含多个 .go 文件，只要这些 .go 文件的头部都使用 package 关键字声明了同一个包。如你所见，Go 语言中 导入的包，必须得用双引号包含，在这里吐槽一下。防止导入的包名和本地的变量发生冲突，比如 path 这个很常用

哈嘿Blog 2020-08-16

GO语言实现标题闪烁效果

因为聊天框是被iframe框进去的页面,所以在聊天框中收到websocket消息以后要把消息发送给父集页面,在父页面中闪烁标题

wfs 2020-08-16

Go语言实现基于websocket浏览器通知功能

当我们使用了websocket技术实现了实时的推送访客来到消息,但是我们只能随时看着页面不离开才能知道有访客来到。现在增加浏览器的通知消息,就可以只要打开着界面,不用一直盯着,干别的事去就可以了,有新消息会从右下角弹出来。notification = ne

蓝色深海 2020-08-16
哈嘿Blog

哈嘿Blog

W3CSchool教程
HTML 教程
CSS 教程
Bootstrap 教程
Javascript 教程
jQuery 教程
后端教程
C 教程
Java 教程
PHP 教程
Python 教程
Go 教程
移动开发
Android 教程
Swift 教程
Kotlin 教程
jQuery Mobile 教程
ionic 教程
关于我们
新闻动态
联系方式
招聘英才
安科实验室
帮助与反馈

安科网(Ancii)，中国第一极客网

Copyright © 2013 - 2019 Ancii.com

京ICP备18063983号 京公网安备11010802014868号