go并发设计模式runner模式

go并发设计模式runner模式

真正运行的程序不可能是单线程运行的，go语言中最值得骄傲的就是CSP模型了，可以说go语言是CSP模型的实现。

假设现在有一个程序需要实现，这个程序有以下要求：

• 程序可以在分配的时间内完成工作，正常终止；
• 程序没有及时完成工作，“自杀”；
• 接收到操作系统发送的中断事件，程序立刻试图清理状态并停止工作

数据类型设计

程序需要在规定时间内完成工作的最简单方法就是使用goroutine和channel，我们需要一个chan用来接收操作完成的信号，完成任务的函数可能有错误信息返回，因此我们这里定义一个错误类型的通道，用来通知什么时候完成任务以及完成任务的错误信息。

complete chan error

任务执行超时的最简单方法就是使用time包提供的After函数，当指定的时间内没有完成任务那么就出发一下超时通道，因为只需要接收超时的信号，因此只需要定义一个单向接收通道即可

timeout <-chan time.Time

当发生系统中断事件时，程序能立刻清理状态然后清理资源并停止工作，因此我们需要一个信号通道来接收操作系统发送的中断信号，这里我们使用signal包提供的Notify函数来注册信号，当操作系统发送信号时，会通过信号通道发送信号

interrupt chan os.Signal

程序最重要的是能够处理任务，用户需要处理多少任务提前是不能确定的，我们需要一个任务列表，这里我们使用一个切片来保存这些任务。

tasks []func(int)

经过上述设计，我们定义一个Runner结构体，用来保存这些通道和任务列表。

// 并且在操作系统发送中断信号时结束这些任务
type Runner struct {// interrupt channel 用来接收操作系统发送的信号interrupt chan os.Signal// complete channel 用来通知任务已经完成complete chan error// timeout channel 用来通知任务已经超时的接收通道timeout <-chan time.Time// tasks 用来保存任务列表tasks []func(int)
}

错误系统设计

错误系统设计，我们希望在任务执行完成或者超时或者操作系统发送的中断信号时返回错误，因此我们定义两个个错误变量，分别用来保存超时错误，中断错误和正常完成错误。

// ErrTimeout 定义一个超时错误, 会在人物执行超时时被返回
var ErrTimeout = errors.New("received timeout")// ErrInterrupt 定义一个中断错误, 会在收到操作系统事件的时候返回
var ErrInterrupt = errors.New("received interrupt")

数据类型说明

Signal

os.Signal 是一个接口类型，是对不同操作系统上捕获的信号的一个抽象接口，用来从操作系统接收中断事件。

// A Signal represents an operating system signal.
// The usual underlying implementation is operating system-dependent:
// on Unix it is syscall.Signal.
type Signal interface {String() stringSignal() // to distinguish from other Stringers
}

Error

error 是一个接口类型，用来表示错误，所有错误类型都实现了error接口，因此我们可以通过error接口来判断错误类型。

Time

time.Time 是一个结构体类型，用来表示一个时间，包含年月日时分秒纳秒等信息。

type Time struct {// wall and ext encode the wall time seconds, wall time nanoseconds,// and optional monotonic clock reading in nanoseconds.//// From high to low bit position, wall encodes a 1-bit flag (hasMonotonic),// a 33-bit seconds field, and a 30-bit wall time nanoseconds field.// The nanoseconds field is in the range [0, 999999999].// If the hasMonotonic bit is 0, then the 33-bit field must be zero// and the full signed 64-bit wall seconds since Jan 1 year 1 is stored in ext.// If the hasMonotonic bit is 1, then the 33-bit field holds a 33-bit// unsigned wall seconds since Jan 1 year 1885, and ext holds a// signed 64-bit monotonic clock reading, nanoseconds since process start.wall uint64ext  int64// loc specifies the Location that should be used to// determine the minute, hour, month, day, and year// that correspond to this Time.// The nil location means UTC.// All UTC times are represented with loc==nil, never loc==&utcLoc.loc *Location
}

方法设计

在go中方法需要示例进行调用，因此我们最后定义一个用来创建Runner实例的New方法，避免用户自行创建实例，导致示例的创建不统一。

名为 New 的工厂函数。这个函数接收一个 time.Duration 类型的值，并返回 Runner 类型的指针。这个函数会创建一个 Runner 类型的值，并初始化每个通道字段。因为 task 字段的零值是 nil，已经满足初始化的要求，所以没有被明确初始化。每个通道字段都有独立的初始化过程

通道 interrupt 被初始化为缓冲区容量为 1 的通道。这可以保证通道至少能接收一个来自语言运行时的 os.Signal 值，确保语言运行时发送这个事件的时候不会被阻塞。如果 goroutine没有准备好接收这个值，这个值就会被丢弃。例如，如果用户反复敲 Ctrl+C 组合键，程序只会在这个通道的缓冲区可用的时候接收事件，其余的所有事件都会被丢弃。

通道 complete 被初始化为无缓冲的通道。当执行任务的 goroutine 完成时，会向这个通道发送一个 error 类型的值或者 nil 值。之后就会等待 main 函数接收这个值。一旦 main 接收了这个 error 值， goroutine 就可以安全地终止了。

最后一个通道 timeout 是用 time 包的 After 函数初始化的。 After 函数返回一个time.Time 类型的通道。语言运行时会在指定的 duration 时间到期之后，向这个通道发送一个 time.Time 的值。

// New 返回一个Runner实例
func New(d time.Duration) *Runner {return &Runner{// 1个缓冲的信号通道interrupt: make(chan os.Signal, 1),// 没有缓冲的信号通道，如果没有接受者那么会阻塞complete: make(chan error),timeout:  time.After(d),}
}

Add 方法用来添加任务，因为需要执行的任务前期并不确定有多少，因此Add接收一个名为tasks的可变参数，可变参数可以接受任意数量的值作为传入参数。这个例子里，这些传入的值必须是一个接收一个整数且什么都
不返回的函数。函数执行时的参数 tasks 是一个存储所有这些传入函数值的切片。

// Add 方法用来添加任务，这个任务是一个接收int类型的ID作为参数的函数
func (r *Runner) Add(tasks ...func(int)) {r.tasks = append(r.tasks, tasks...)
}

run 方法会迭代 tasks 切片，并按顺序执行每个函数

func (r *Runner) run() error {for id, task := range r.tasks {if r.gotInterrupt() {return ErrInterrupt}// 执行注册的任务task(id)}return nil
}

gotInterrupt 展示了带 default 分支的 select 语句的经典用法。代码试图从 interrupt 通道去接收信号。一般来说， select 语句在没有任何要接收的数据时会阻塞，不过有了 default 分支就不会阻塞了。 default 分支会将接收 interrupt 通道的阻塞调用转变为非阻塞的。如果 interrupt 通道有中断信号需要接收，就会接收并处理这个中断。如果没有需要接收的信号，就会执行 default 分支。当收到中断信号后，代码会通过调用 Stop 方法来停止接收之后的所有事件。之后函数返回 true。如果没有收到中断信号，在第 99 行该方法会返回 false。本质上，gotInterrupt 方法会让 goroutine 检查中断信号，如果没有发出中断信号，就继续处理工作。

// gotInterrupt 检查是否接收到中断信号
func (r *Runner) gotInterrupt() bool {select {// 如果有中断信号那么返回truecase <-r.interrupt:// 接收到中断信号，停止后续再接收到中断信号signal.Stop(r.interrupt)return true// 没有终端信号返回false，继续执行default:return false}
}

一切步骤都执行完了，现在开始执行任务

// Start 方法用来开始执行任务，并监视通道事件
func (r *Runner) Start() error {// 我们希望接收所有中断信号signal.Notify(r.interrupt, os.Interrupt)// 异步执行任务go func() {r.complete <- r.run()}()select {// 当任务处理完成时该通道会返回case err := <-r.complete:return err// 当任务处理程序运行超时时发出信号case <-r.timeout:return ErrTimeout}
}

将以上代码全部都整合到runner.go文件中

// Package runner 处理任务的运行和声明周期管理
package runnerimport ("errors""os""os/signal""time"
)// Runner 在给定的超时时间内执行一组任务
// 并且在操作系统发送中断信号时结束这些任务
type Runner struct {// interrupt channel 用来接收操作系统发送的信号interrupt chan os.Signal// complete channel 用来通知任务已经完成complete chan error// timeout channel 用来通知任务已经超时timeout <-chan time.Time// tasks 用来保存任务列表tasks []func(int)
}// ErrTimeout 定义一个超时错误, 会在人物执行超时时被返回
var ErrTimeout = errors.New("received timeout")// ErrInterrupt 定义一个中断错误, 会在收到操作系统事件的时候返回
var ErrInterrupt = errors.New("received interrupt")// New 返回一个Runner实例
func New(d time.Duration) *Runner {return &Runner{// 1个缓冲的信号通道interrupt: make(chan os.Signal, 1),// 没有缓冲的信号通道，如果没有接受者那么会阻塞complete: make(chan error),timeout:  time.After(d),}
}// Add 方法用来添加任务，这个任务是一个接收int类型的ID作为参数的函数
func (r *Runner) Add(tasks ...func(int)) {r.tasks = append(r.tasks, tasks...)
}// Start 方法用来开始执行任务，并监视通道事件
func (r *Runner) Start() error {// 我们希望接收所有中断信号signal.Notify(r.interrupt, os.Interrupt)// 异步执行任务go func() {r.complete <- r.run()}()select {// 当任务处理完成时该通道会返回case err := <-r.complete:return err// 当任务处理程序运行超时时发出信号case <-r.timeout:return ErrTimeout}}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 {// 如果有中断信号那么返回truecase <-r.interrupt:// 接收到中断信号，停止后续再接收到中断信号signal.Stop(r.interrupt)return true// 没有终端信号返回false，继续执行default:return false}
}

在main.go中进行调用

package mainimport ("log""os""time""code/runner"
)// timeout 定义程序执行超时时间，如果超过这个时间还没执行完成会失败退出.
const timeout = 3 * time.Second// 主函数入口
func main() {log.Println("Starting work.")// 调用New创建 runner对象.r := runner.New(timeout)// 向任务队列中添加需要顺序执行的任务r.Add(createTask(), createTask(), createTask())// Run 执行人物，并按照返回错误处理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 返回一个入参为int的函数
func createTask() func(int) {return func(id int) {log.Printf("Processor - Task #%d.", id)time.Sleep(time.Duration(id) * time.Second)}
}

源码已经放到gitee需要的自行下载：
https://gitee.com/andrewgithub/note_lab/blob/main/example/go/concurrent_mode/runner/runner.go