Go語言實戰–runner

代碼

runner.go

// Example is provided with help by Gabriel Aszalos. // Package runner manages the running and lifetime of a process. package runnerimport ("errors""os""os/signal""time" )// Runner runs a set of tasks within a given timeout and can be // shut down on an operating system interrupt. type Runner struct {// interrupt channel reports a signal from the// operating system.interrupt chan os.Signal// complete channel reports that processing is done.complete chan error// timeout reports that time has run out.timeout <-chan time.Time// tasks holds a set of functions that are executed// synchronously in index order.tasks []func(int) }// ErrTimeout is returned when a value is received on the timeout channel. var ErrTimeout = errors.New("received timeout")// ErrInterrupt is returned when an event from the OS is received. var ErrInterrupt = errors.New("received interrupt")// New returns a new ready-to-use Runner. func New(d time.Duration) *Runner {return &Runner{interrupt: make(chan os.Signal, 1),complete: make(chan error),timeout: time.After(d),} }// Add attaches tasks to the Runner. A task is a function that // takes an int ID. func (r *Runner) Add(tasks ...func(int)) {r.tasks = append(r.tasks, tasks...) }// Start runs all tasks and monitors channel events. func (r *Runner) Start() error {// We want to receive all interrupt based signals.signal.Notify(r.interrupt, os.Interrupt)// Run the different tasks on a different goroutine.go func() {r.complete <- r.run()}()select {// Signaled when processing is done.case err := <-r.complete:return err// Signaled when we run out of time.case <-r.timeout:return ErrTimeout} }// run executes each registered task. func (r *Runner) run() error {for id, task := range r.tasks {// Check for an interrupt signal from the OS.if r.gotInterrupt() {return ErrInterrupt}// Execute the registered task.task(id)}return nil }// gotInterrupt verifies if the interrupt signal has been issued. func (r *Runner) gotInterrupt() bool {select {// Signaled when an interrupt event is sent.case <-r.interrupt:// Stop receiving any further signals.signal.Stop(r.interrupt)return true// Continue running as normal.default:return false} }

main.go

// This sample program demonstrates how to use a channel to // monitor the amount of time the program is running and terminate // the program if it runs too long. package mainimport ("log""os""time""github.com/goinaction/code/chapter7/patterns/runner" )// timeout is the number of second the program has to finish. const timeout = 3 * time.Second// main is the entry point for the program. func main() {log.Println("Starting work.")// Create a new timer value for this run.r := runner.New(timeout)// Add the tasks to be run.r.Add(createTask(), createTask(), createTask())// Run the tasks and handle the result.if err := r.Start(); err != nil {switch err {case runner.ErrTimeout:log.Println("Terminating due to timeout.")os.Exit(1)case runner.ErrInterrupt:log.Println("Terminating due to interrupt.")os.Exit(2)}}log.Println("Process ended.") }// createTask returns an example task that sleeps for the specified // number of seconds based on the id. func createTask() func(int) {return func(id int) {log.Printf("Processor - Task #%d.", id)time.Sleep(time.Duration(id) * time.Second)} }

代碼分析

runner包：

使用通道監視程序的執行時間，當程序運行時間太長時，可以使用runner包中止程序。

type Runner struct {interrupt chan os.Signalcomplete chan errortimeout <-chan time.Timetasks []func(int) }

Runner結構體：

定義了

1 通過channel接收來自操作系統發送的信號

2 任務完成的信號

3 超時信號

4 任務切片

New

func New(d time.Duration) *Runner {return &Runner{interrupt: make(chan os.Signal, 1),complete: make(chan error),timeout: time.After(d),} }

通過New返回一個新的Runner：

complete:

當執行任務的goroutine完成時，會向這個通道發送一個error類型的值或者nil值。之后就會等待main函數接收這個值。一旦main接收了這個error值，goroutine就可以安全地終止了

time.After(d)：

func After(d Duration) <-chan Time {return NewTimer(d).C }

Add

func (r *Runner) Add(tasks ...func(int)) {r.tasks = append(r.tasks, tasks...) }

Add方法，這個方法接收一個名為 tasks 的可變參數。可變參數可以接受任意數量的值作為傳入參數。這個例子里，這些傳入的值必須是一個接收一個整數且什么都不返回的函數。函數執行時的 參數tasks是一個存儲所有這些傳入函數值的切片

run

func (r *Runner) run() error {for id, task := range r.tasks {if r.gotInterrupt() {return ErrInterrupt}task(id)}return nil }

檢查操作系統有無中斷信號，沒有則繼續執行

gotInterrupt

func (r *Runner) gotInterrupt() bool {select {case <-r.interrupt:signal.Stop(r.interrupt)return truedefault:return false} }

其中signal.Stop(r.interrupt)：

func Stop(c chan<- os.Signal)

Stop函數讓signal包停止向c轉發信號。它會取消之前使用c調用的所有Notify的效果。當Stop返回后，會保證c不再接收到任何信號。

start

func (r *Runner) Start() error {signal.Notify(r.interrupt, os.Interrupt)go func() {r.complete <- r.run()}()select {case err := <-r.complete:return errcase <-r.timeout:return ErrTimeout} }

實現了程序的主流程：

func Notify(c chan<- os.Signal, sig ...os.Signal)

Notify函數讓signal包將輸入信號轉發到c。如果沒有列出要傳遞的信號，會將所有輸入信號傳遞到c；否則只傳遞列出的輸入信號。

signal包不會為了向c發送信息而阻塞（就是說如果發送時c阻塞了，signal包會直接放棄）：調用者應該保證c有足夠的緩存空間可以跟上期望的信號頻率。對使用單一信號用于通知的通道，緩存為1就足夠了。

main.go

r.Add(createTask(), createTask(), createTask()) func createTask() func(int) {return func(id int) {log.Printf("Processor - Task #%d.", id)time.Sleep(time.Duration(id) * time.Second)} }

總結

以上是生活随笔為你收集整理的Go语言实战--学习笔记--runner的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。