Go語言同步與異步執(zhí)行多個任務(wù)封裝詳解(Runner和RunnerAsync)
更新時間：2018年01月12日 11:00:42 作者：雪山飛豬
這篇文章主要給大家介紹了關(guān)于Go語言同步與異步執(zhí)行多個任務(wù)封裝(Runner和RunnerAsync)的相關(guān)資料，文中通過示例代碼介紹的非常詳細(xì)，對大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價值，需要的朋友們下面隨著小編來一起學(xué)習(xí)學(xué)習(xí)吧。
前言
同步適合多個連續(xù)執(zhí)行的，每一步的執(zhí)行依賴于上一步操作，異步執(zhí)行則和任務(wù)執(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í)行任務(wù)
type Runner struct {
//操作系統(tǒng)的信號檢測
interrupt chan os.Signal
//記錄執(zhí)行完成的狀態(tài)
complete chan error
//超時檢測
timeout <-chan time.Time
//保存所有要執(zhí)行的任務(wù),順序執(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{},
}
}
//添加一個任務(wù)
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í)行任務(wù)
go func() {
this.complete <- this.Run()
}()
select {
//返回執(zhí)行結(jié)果
case err := <-this.complete:
return err
//超時返回
case <-this.timeout:
return ErrTimeout
}
}
//異步執(zhí)行所有的任務(wù)
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í)行任務(wù)
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添加一個任務(wù)，任務(wù)為接收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對象，設(shè)置超時時間
runner := NewRunnerAsync(8 * time.Second)
//添加運(yùn)行的任務(wù)
runner.Add(
createTaskAsync(),
createTaskAsync(),
createTaskAsync(),
createTaskAsync(),
createTaskAsync(),
createTaskAsync(),
createTaskAsync(),
createTaskAsync(),
createTaskAsync(),
createTaskAsync(),
createTaskAsync(),
createTaskAsync(),
createTaskAsync(),
)
fmt.Println("同步執(zhí)行任務(wù)")
//開始執(zhí)行任務(wù)
if err := runner.Start(); err != nil {
switch err {
case ErrTimeout:
fmt.Println("執(zhí)行超時")
os.Exit(1)
case ErrInterrupt:
fmt.Println("任務(wù)被中斷")
os.Exit(2)
}
}
t.Log("執(zhí)行結(jié)束")
}
//創(chuàng)建要執(zhí)行的任務(wù)
func createTaskAsync() func(id int) {
return func(id int) {
fmt.Printf("正在執(zhí)行%v個任務(wù)\n", id)
//模擬任務(wù)執(zhí)行,sleep兩秒
//time.Sleep(1 * time.Second)
}
}
執(zhí)行結(jié)果
同步執(zhí)行任務(wù) 正在執(zhí)行0個任務(wù) 正在執(zhí)行1個任務(wù) 正在執(zhí)行2個任務(wù) 正在執(zhí)行3個任務(wù) 正在執(zhí)行4個任務(wù) 正在執(zhí)行5個任務(wù) 正在執(zhí)行6個任務(wù) 正在執(zhí)行7個任務(wù) 正在執(zhí)行8個任務(wù) 正在執(zhí)行9個任務(wù) 正在執(zhí)行10個任務(wù) 正在執(zhí)行11個任務(wù) 正在執(zhí)行12個任務(wù) runnerAsync_test.go:49: 執(zhí)行結(jié)束
異步執(zhí)行類Runner
支持返回超時檢測，系統(tǒng)中斷檢測
實現(xiàn)代碼如下
package task
import (
"os"
"time"
"os/signal"
"sync"
)
//異步執(zhí)行任務(wù)
type Runner struct {
//操作系統(tǒng)的信號檢測
interrupt chan os.Signal
//記錄執(zhí)行完成的狀態(tài)
complete chan error
//超時檢測
timeout <-chan time.Time
//保存所有要執(zhí)行的任務(wù),順序執(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{},
}
}
//添加一個任務(wù)
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í)行任務(wù)
go func() {
this.complete <- this.Run()
}()
select {
//返回執(zhí)行結(jié)果
case err := <-this.complete:
return err
//超時返回
case <-this.timeout:
return ErrTimeout
}
}
//異步執(zhí)行所有的任務(wù)
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í)行任務(wù)
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添加一個任務(wù)，任務(wù)為接收int類型，返回類型error的一個閉包
Start開始執(zhí)行傷，返回一個error類型，nil為執(zhí)行完畢, ErrTimeout代表執(zhí)行超時，ErrInterrupt代表執(zhí)行被中斷(類似Ctrl + C操作)
getErrs獲取所有的任務(wù)執(zhí)行結(jié)果
測試示例代碼
package task
import (
"testing"
"time"
"fmt"
"os"
"runtime"
)
func TestRunner_Start(t *testing.T) {
//開啟多核心
runtime.GOMAXPROCS(runtime.NumCPU())
//創(chuàng)建runner對象，設(shè)置超時時間
runner := NewRunner(18 * time.Second)
//添加運(yùn)行的任務(wù)
runner.Add(
createTask(),
createTask(),
createTask(),
createTask(),
createTask(),
createTask(),
createTask(),
createTask(),
createTask(),
createTask(),
createTask(),
createTask(),
createTask(),
createTask(),
)
fmt.Println("異步執(zhí)行任務(wù)")
//開始執(zhí)行任務(wù)
if err := runner.Start(); err != nil {
switch err {
case ErrTimeout:
fmt.Println("執(zhí)行超時")
os.Exit(1)
case ErrInterrupt:
fmt.Println("任務(wù)被中斷")
os.Exit(2)
}
}
t.Log("執(zhí)行結(jié)束")
t.Log(runner.GetErrs())
}
//創(chuàng)建要執(zhí)行的任務(wù)
func createTask() func(id int) error {
return func(id int) error {
fmt.Printf("正在執(zhí)行%v個任務(wù)\n", id)
//模擬任務(wù)執(zhí)行,sleep
//time.Sleep(1 * time.Second)
return nil
}
}
執(zhí)行結(jié)果
異步執(zhí)行任務(wù) 正在執(zhí)行2個任務(wù) 正在執(zhí)行1個任務(wù) 正在執(zhí)行4個任務(wù) 正在執(zhí)行3個任務(wù) 正在執(zhí)行6個任務(wù) 正在執(zhí)行5個任務(wù) 正在執(zhí)行9個任務(wù) 正在執(zhí)行7個任務(wù) 正在執(zhí)行10個任務(wù) 正在執(zhí)行13個任務(wù) 正在執(zhí)行8個任務(wù) 正在執(zhí)行11個任務(wù) 正在執(zhí)行12個任務(wù) 正在執(zhí)行0個任務(wù) 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)容對大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價值，如果有疑問大家可以留言交流，謝謝大家對腳本之家的支持。
相關(guān)文章
Golang設(shè)計模式工廠模式實戰(zhàn)寫法示例詳解
這篇文章主要為大家介紹了Golang 工廠模式實戰(zhàn)寫法示例詳解，有需要的朋友可以借鑒參考下，希望能夠有所幫助，祝大家多多進(jìn)步，早日升職加薪2022-08-08
這篇文章主要介紹了Golang通脈之?dāng)?shù)據(jù)類型，在編程語言中標(biāo)識符就是定義的具有某種意義的詞，比如變量名、常量名、函數(shù)名等等，Go語言中標(biāo)識符允許由字母數(shù)字和_(下劃線）組成，并且只能以字母和_開頭，更詳細(xì)內(nèi)容請看下面文章吧2021-10-10
Go錯誤處理涵蓋接口、創(chuàng)建方式、檢查模式、包裝、最佳實踐及工具庫,強(qiáng)調(diào)顯式檢查、簡單可預(yù)測和錯誤即值理念,提升代碼清晰度和可靠性2025-07-07
這篇文章主要介紹了golang不到30行代碼實現(xiàn)依賴注入的方法,小編覺得挺不錯的，現(xiàn)在分享給大家，也給大家做個參考。一起跟隨小編過來看看吧2018-07-07
適配器是一種結(jié)構(gòu)型設(shè)計模式， 它能使不兼容的對象能夠相互合作，可擔(dān)任兩個對象間的封裝器， 它會接收對于一個對象的調(diào)用， 并將其轉(zhuǎn)換為另一個對象可識別的格式和接口，本文將通過代碼示例詳細(xì)給大家介紹Golang的適配器模式2023-06-06
這篇文章主要介紹了在Golang中使用Redis的方法示例,小編覺得挺不錯的，現(xiàn)在分享給大家，也給大家做個參考。一起跟隨小編過來看看吧2018-06-06
GO語言入門學(xué)習(xí)之基本數(shù)據(jù)類型字符串
字符串在Go語言中以原生數(shù)據(jù)類型出現(xiàn),使用字符串就像使用其他原生數(shù)據(jù)類型(int、bool、float32、float64 等)一樣,下面這篇文章主要給大家介紹了關(guān)于GO語言入門學(xué)習(xí)之基本數(shù)據(jù)類型字符串的相關(guān)資料,需要的朋友可以參考下2022-04-04
這篇文章主要介紹了詳解Go-JWT-RESTful身份認(rèn)證教程，文中通過示例代碼介紹的非常詳細(xì)，對大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價值，需要的朋友們下面隨著小編來一起學(xué)習(xí)學(xué)習(xí)吧2019-09-09
這篇文章主要介紹了golang代碼檢測工具之goimports使用,具有很好的參考價值,希望對大家有所幫助,如有錯誤或未考慮完全的地方,望不吝賜教2024-01-01
Go語言的log包提供了用于記錄日志的基本功能,相比fmt包,增加了時間戳、日志級別等日志管理功能,非常適合用于調(diào)試和記錄運(yùn)行信息,本文呢介紹go語言中的log 包,感興趣的朋友一起看看吧2024-11-11