Go語言同步與異步執(zhí)行多個任務封裝詳解(Runner和RunnerAsync)
更新時間：2018年01月12日 11:00:42 作者：雪山飛豬
這篇文章主要給大家介紹了關(guān)于Go語言同步與異步執(zhí)行多個任務封裝(Runner和RunnerAsync)的相關(guān)資料，文中通過示例代碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值，需要的朋友們下面隨著小編來一起學習學習吧。
前言
同步適合多個連續(xù)執(zhí)行的，每一步的執(zhí)行依賴于上一步操作，異步執(zhí)行則和任務執(zhí)行順序無關(guān)(如從10個站點抓取數(shù)據(jù))
同步執(zhí)行類RunnerAsync
支持返回超時檢測，系統(tǒng)中斷檢測
錯誤常量定義
//超時錯誤
var ErrTimeout = errors.New("received timeout")
//操作系統(tǒng)系統(tǒng)中斷錯誤
var ErrInterrupt = errors.New("received interrupt")
實現(xiàn)代碼如下
package task
import (
"os"
"time"
"os/signal"
"sync"
)
//異步執(zhí)行任務
type Runner struct {
//操作系統(tǒng)的信號檢測
interrupt chan os.Signal
//記錄執(zhí)行完成的狀態(tài)
complete chan error
//超時檢測
timeout <-chan time.Time
//保存所有要執(zhí)行的任務,順序執(zhí)行
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，監(jiān)聽錯誤信息
func (this *Runner) Start() error {
//接收操作系統(tǒng)信號
signal.Notify(this.interrupt, os.Interrupt)
//并發(fā)執(zhí)行任務
go func() {
this.complete <- this.Run()
}()
select {
//返回執(zhí)行結(jié)果
case err := <-this.complete:
return err
//超時返回
case <-this.timeout:
return ErrTimeout
}
}
//異步執(zhí)行所有的任務
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()
//執(zhí)行任務
err := task(id)
//加鎖保存到結(jié)果集中
this.errs = append(this.errs, err)
this.lock.Unlock()
this.waitGroup.Done()
}(id)
}
this.waitGroup.Wait()
return nil
}
//判斷是否接收到操作系統(tǒng)中斷信號
func (this *Runner) gotInterrupt() bool {
select {
case <-this.interrupt:
//停止接收別的信號
signal.Stop(this.interrupt)
return true
//正常執(zhí)行
default:
return false
}
}
//獲取執(zhí)行完的error
func (this *Runner) GetErrs() []error {
return this.errs
}
使用方法
Add添加一個任務，任務為接收int類型的一個閉包
Start開始執(zhí)行傷，返回一個error類型，nil為執(zhí)行完畢, ErrTimeout代表執(zhí)行超時，ErrInterrupt代表執(zhí)行被中斷(類似Ctrl + C操作)
測試示例代碼
package task
import (
"testing"
"time"
"fmt"
"os"
"runtime"
)
func TestRunnerAsync_Start(t *testing.T) {
//開啟多核
runtime.GOMAXPROCS(runtime.NumCPU())
//創(chuàng)建runner對象，設置超時時間
runner := NewRunnerAsync(8 * time.Second)
//添加運行的任務
runner.Add(
createTaskAsync(),
createTaskAsync(),
createTaskAsync(),
createTaskAsync(),
createTaskAsync(),
createTaskAsync(),
createTaskAsync(),
createTaskAsync(),
createTaskAsync(),
createTaskAsync(),
createTaskAsync(),
createTaskAsync(),
createTaskAsync(),
)
fmt.Println("同步執(zhí)行任務")
//開始執(zhí)行任務
if err := runner.Start(); err != nil {
switch err {
case ErrTimeout:
fmt.Println("執(zhí)行超時")
os.Exit(1)
case ErrInterrupt:
fmt.Println("任務被中斷")
os.Exit(2)
}
}
t.Log("執(zhí)行結(jié)束")
}
//創(chuàng)建要執(zhí)行的任務
func createTaskAsync() func(id int) {
return func(id int) {
fmt.Printf("正在執(zhí)行%v個任務\n", id)
//模擬任務執(zhí)行,sleep兩秒
//time.Sleep(1 * time.Second)
}
}
執(zhí)行結(jié)果
同步執(zhí)行任務 正在執(zhí)行0個任務 正在執(zhí)行1個任務 正在執(zhí)行2個任務 正在執(zhí)行3個任務 正在執(zhí)行4個任務 正在執(zhí)行5個任務 正在執(zhí)行6個任務 正在執(zhí)行7個任務 正在執(zhí)行8個任務 正在執(zhí)行9個任務 正在執(zhí)行10個任務 正在執(zhí)行11個任務 正在執(zhí)行12個任務 runnerAsync_test.go:49: 執(zhí)行結(jié)束
異步執(zhí)行類Runner
支持返回超時檢測，系統(tǒng)中斷檢測
實現(xiàn)代碼如下
package task
import (
"os"
"time"
"os/signal"
"sync"
)
//異步執(zhí)行任務
type Runner struct {
//操作系統(tǒng)的信號檢測
interrupt chan os.Signal
//記錄執(zhí)行完成的狀態(tài)
complete chan error
//超時檢測
timeout <-chan time.Time
//保存所有要執(zhí)行的任務,順序執(zhí)行
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，監(jiān)聽錯誤信息
func (this *Runner) Start() error {
//接收操作系統(tǒng)信號
signal.Notify(this.interrupt, os.Interrupt)
//并發(fā)執(zhí)行任務
go func() {
this.complete <- this.Run()
}()
select {
//返回執(zhí)行結(jié)果
case err := <-this.complete:
return err
//超時返回
case <-this.timeout:
return ErrTimeout
}
}
//異步執(zhí)行所有的任務
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()
//執(zhí)行任務
err := task(id)
//加鎖保存到結(jié)果集中
this.errs = append(this.errs, err)
this.lock.Unlock()
this.waitGroup.Done()
}(id)
}
this.waitGroup.Wait()
return nil
}
//判斷是否接收到操作系統(tǒng)中斷信號
func (this *Runner) gotInterrupt() bool {
select {
case <-this.interrupt:
//停止接收別的信號
signal.Stop(this.interrupt)
return true
//正常執(zhí)行
default:
return false
}
}
//獲取執(zhí)行完的error
func (this *Runner) GetErrs() []error {
return this.errs
}
使用方法
Add添加一個任務，任務為接收int類型，返回類型error的一個閉包
Start開始執(zhí)行傷，返回一個error類型，nil為執(zhí)行完畢, ErrTimeout代表執(zhí)行超時，ErrInterrupt代表執(zhí)行被中斷(類似Ctrl + C操作)
getErrs獲取所有的任務執(zhí)行結(jié)果
測試示例代碼
package task
import (
"testing"
"time"
"fmt"
"os"
"runtime"
)
func TestRunner_Start(t *testing.T) {
//開啟多核心
runtime.GOMAXPROCS(runtime.NumCPU())
//創(chuàng)建runner對象，設置超時時間
runner := NewRunner(18 * time.Second)
//添加運行的任務
runner.Add(
createTask(),
createTask(),
createTask(),
createTask(),
createTask(),
createTask(),
createTask(),
createTask(),
createTask(),
createTask(),
createTask(),
createTask(),
createTask(),
createTask(),
)
fmt.Println("異步執(zhí)行任務")
//開始執(zhí)行任務
if err := runner.Start(); err != nil {
switch err {
case ErrTimeout:
fmt.Println("執(zhí)行超時")
os.Exit(1)
case ErrInterrupt:
fmt.Println("任務被中斷")
os.Exit(2)
}
}
t.Log("執(zhí)行結(jié)束")
t.Log(runner.GetErrs())
}
//創(chuàng)建要執(zhí)行的任務
func createTask() func(id int) error {
return func(id int) error {
fmt.Printf("正在執(zhí)行%v個任務\n", id)
//模擬任務執(zhí)行,sleep
//time.Sleep(1 * time.Second)
return nil
}
}
執(zhí)行結(jié)果
異步執(zhí)行任務 正在執(zhí)行2個任務 正在執(zhí)行1個任務 正在執(zhí)行4個任務 正在執(zhí)行3個任務 正在執(zhí)行6個任務 正在執(zhí)行5個任務 正在執(zhí)行9個任務 正在執(zhí)行7個任務 正在執(zhí)行10個任務 正在執(zhí)行13個任務 正在執(zhí)行8個任務 正在執(zhí)行11個任務 正在執(zhí)行12個任務 正在執(zhí)行0個任務 runner_test.go:49: 執(zhí)行結(jié)束 runner_test.go:51: [<nil> <nil> <nil> <nil> <nil> <nil> <nil> <nil> <nil> <nil> <nil> <nil> <nil> <nil>]
總結(jié)
以上就是這篇文章的全部內(nèi)容了，希望本文的內(nèi)容對大家的學習或者工作具有一定的參考學習價值，如果有疑問大家可以留言交流，謝謝大家對腳本之家的支持。
相關(guān)文章
JWT,全稱 JSON Web Token,是一種開放標準（RFC 7519）,用于安全地在雙方之間傳遞信息,這篇文章主要介紹了golang 在Gin框架中使用JWT鑒權(quán),需要的朋友可以參考下2024-07-07
內(nèi)存逃逸分析是go的編譯器在編譯期間，根據(jù)變量的類型和作用域，確定變量是堆上還是棧上。本文將通過示例淺析一下Golang中的內(nèi)存逃逸，需要的可以了解一下2022-10-10
gin解析json格式的數(shù)據(jù)出錯的處理方案
這篇文章主要介紹了gin解析json格式的數(shù)據(jù)出錯的處理方案，具有很好的參考價值，希望對大家有所幫助。一起跟隨小編過來看看吧2021-03-03
這篇文章主要介紹了go語言中的Carbon庫時間處理,本文給大家介紹的非常詳細，對大家的學習或工作具有一定的參考借鑒價值，需要的朋友可以參考下2021-02-02
在Go語言中，將interface設計為引用類型是為了實現(xiàn)更靈活、更動態(tài)的類型系統(tǒng)，這篇文章主要介紹了深度解析Golang中為什么interface是引用類型,需要的朋友可以參考下2024-01-01
這篇文章主要介紹了Go標準庫http與fasthttp服務端性能比較,本文給大家介紹的非常詳細，對大家的學習或工作具有一定的參考借鑒價值，需要的朋友可以參考下2022-06-06
今天小編就為大家分享一篇golang解析html網(wǎng)頁的方法，具有很好的參考價值，希望對大家有所幫助。一起跟隨小編過來看看吧2019-08-08
GOLANG使用Context實現(xiàn)傳值、超時和取消的方法
這篇文章主要介紹了GOLANG使用Context實現(xiàn)傳值、超時和取消的方法，小編覺得挺不錯的，現(xiàn)在分享給大家，也給大家做個參考。一起跟隨小編過來看看吧2019-01-01
這篇文章主要為大家詳細介紹Go語言中函數(shù)的定義與使用，文中的示例代碼講解詳細，對我們學習Go語言有一定幫助，需要的可以參考一下2022-04-04
這篇文章主要介紹了golang之反射和斷言的具體使用，文中通過示例代碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值，需要的朋友們下面隨著小編來一起學習學習吧2019-10-10