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

作者：隐身勇者_641 | 来源：互联网 | 2022-11-18 12:53

这篇文章主要给大家介绍了关于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: [             ]

总结

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

推荐阅读

python
golang 解析磁力链为 torrent 相关的信息

其实通过http请求已经获得了种子的信息了，但是传播存储种子好像是违法的，所以就存储些描述信息吧。之前python跑的太慢了。这个go并发不知道写的有没有问题？！packag ... [详细]

蜡笔小新 2023-10-11 04:54:46
hook
深入探讨C++中的GCD函数与队列

在iOS开发中，多线程技术的应用非常广泛，能够高效地执行多个调度任务。本文将重点介绍GCD（Grand Central Dispatch）在多线程开发中的应用，包括其函数和队列的实现细节。 ... [详细]

蜡笔小新 2024-11-16 14:59:50
command
深入理解Java多线程与并发机制

本文探讨了Java多线程和并发机制的核心概念，包括多线程类的分类、执行器框架、并发容器及控制工具。通过详细解析这些组件，帮助开发者更好地理解和应用多线程技术。 ... [详细]

蜡笔小新 2024-11-16 07:30:39
case
Android异步处理系列文章四篇之三

Android异步处理一：使用Thread+Handler实现非UI线程更新UI界面Android异步处理二：使用AsyncTask异步更新UI界面Android异步处理三：Handler+Loope ... [详细]

蜡笔小新 2024-11-15 19:09:29
js
Go Echo 框架入门指南【1】

本文介绍了 Go 语言中的高性能、可扩展、轻量级 Web 框架 Echo。Echo 框架简单易用，仅需几行代码即可启动一个高性能 HTTP 服务。 ... [详细]

蜡笔小新 2024-11-14 18:30:58
schema
Spring – Bean Life Cycle

Spring – Bean Life Cycle ... [详细]

蜡笔小新 2024-11-13 13:24:40
case
Go (Golang) 语言Golang 定时器Timer和Ticker、time.AfterFunc、time.NewTicker()实例

文章目录Golang定时器Timer和Tickertime.Timertime.NewTimer()实例time.AfterFunctime.Tickertime.NewTicke ... [详细]

蜡笔小新 2024-11-12 09:39:10
list
如何更有效地提升对支持部门的协助与支撑？ - Enhancing Support for the Support Department: Strategies and Best Practices

尽管我们尽最大努力，任何软件开发过程中都难免会出现缺陷。为了更有效地提升对支持部门的协助与支撑，本文探讨了多种策略和最佳实践，旨在通过改进沟通、增强培训和支持流程来减少这些缺陷的影响，并提高整体服务质量和客户满意度。 ... [详细]

蜡笔小新 2024-11-07 06:55:33
list
AIX编程挑战赛：AIX正方形问题的算法解析与Java代码实现

在昨晚的阅读中，我注意到了CSDN博主西部阿呆-小草屋发表的一篇文章《AIX程序设计大赛——AIX正方形问题》。该文详细阐述了AIX正方形问题的背景，并提供了一种基于Java语言的解决方案。本文将深入解析这一算法的核心思想，并展示具体的Java代码实现，旨在为参赛者和编程爱好者提供有价值的参考。 ... [详细]

蜡笔小新 2024-11-06 16:47:06
case
深入解析 Golang 中 Context 的功能与应用

本文详细探讨了 Golang 中 Context 的核心功能及其应用场景，通过深入解析其工作机制，帮助读者更好地理解和运用这一重要特性，对于提升代码质量和项目开发效率具有重要的参考价值。 ... [详细]

蜡笔小新 2024-11-06 16:35:57
ascii
Netty框架中运用Protobuf实现高效通信协议

在Netty框架中，通过引入Protobuf来实现高效的通信协议。为了使用Protobuf，需要先准备好环境，包括下载并安装Protobuf的代码生成器`protoc`以及相应的源码包。具体资源可从官方下载页面获取，确保版本兼容性以充分发挥其性能优势。此外，配置好开发环境后，可以通过定义`.proto`文件来自动生成Java类，从而简化数据序列化和反序列化的操作，提高通信效率。 ... [详细]

蜡笔小新 2024-11-05 17:06:20
fetch
深入解析Zebra中的线程机制及其应用

本文详细探讨了Zebra路由软件中的线程机制及其实际应用。通过对Zebra线程模型的深入分析，揭示了其在高效处理网络路由任务中的关键作用。文章还介绍了线程同步与通信机制，以及如何通过优化线程管理提升系统性能。此外，结合具体应用场景，展示了Zebra线程机制在复杂网络环境下的优势和灵活性。 ... [详细]

蜡笔小新 2024-11-04 19:18:15
schema
利用Java实现WebSocket实时消息推送技术

本文探讨了利用Java实现WebSocket实时消息推送技术的方法。与传统的轮询、长连接或短连接等方案相比，WebSocket提供了一种更为高效和低延迟的双向通信机制。通过建立持久连接，服务器能够主动向客户端推送数据，从而实现真正的实时消息传递。此外，本文还介绍了WebSocket在实际应用中的优势和应用场景，并提供了详细的实现步骤和技术细节。 ... [详细]

蜡笔小新 2024-11-04 15:50:18
schema
Understanding the Concept and Usage of Null in Java Programming

https:www.hollischuang.comarchives74 对于Java程序员来说，null是令人头痛的东西。时常会受到空指针异常（NPE ... [详细]

蜡笔小新 2024-10-22 12:45:48
fetch
在Oracle中执行动态SQL的几种方法,sql server 自定义函数的使用

在Oracle中执行动态SQL的几种方法在一般的sql操作中，sql语句基本上都是固定的，如：SELECTt.empno,t.ename FROMscott.emptWHEREt. ... [详细]

蜡笔小新 2024-09-30 13:57:51

隐身勇者_641

这个家伙很懒，什么也没留下！

Tags | 热门标签

RankList | 热门文章