Go语言实战--学习笔记--runner

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)}
}