Go語言同步與異步執(zhí)行多個任務封裝詳解(Runner和RunnerAsync)
前言
同步適合多個連續(xù)執(zhí)行的,每一步的執(zhí)行依賴于上一步操作,異步執(zhí)行則和任務執(zhí)行順序無關(如從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í)行結果
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)
//加鎖保存到結果集中
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í)行結束")
}
//創(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í)行結果
同步執(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í)行結束
異步執(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í)行結果
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)
//加鎖保存到結果集中
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í)行結果
測試示例代碼
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í)行結束")
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í)行結果
異步執(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í)行結束 runner_test.go:51: [<nil> <nil> <nil> <nil> <nil> <nil> <nil> <nil> <nil> <nil> <nil> <nil> <nil> <nil>]
總結
以上就是這篇文章的全部內容了,希望本文的內容對大家的學習或者工作具有一定的參考學習價值,如果有疑問大家可以留言交流,謝謝大家對腳本之家的支持。

