网站大气是什么意思,wordpress博客批量发布,厦门网络推广外包多少钱,沈阳网站建站公司go并发设计模式runner模式 真正运行的程序不可能是单线程运行的#xff0c;go语言中最值得骄傲的就是CSP模型了#xff0c;可以说go语言是CSP模型的实现。
假设现在有一个程序需要实现#xff0c;这个程序有以下要求#xff1a; 程序可以在分配的时间内完成工作#xff0…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 locnil, never locutcLoc.loc *Location
}方法设计
在go中方法需要示例进行调用因此我们最后定义一个用来创建Runner实例的New方法避免用户自行创建实例导致示例的创建不统一。
名为 New 的工厂函数。这个函数接收一个 time.Duration 类型的值并返回 Runner 类型的指针。这个函数会创建一个 Runner 类型的值并初始化每个通道字段。因为 task 字段的零值是 nil已经满足初始化的要求所以没有被明确初始化。每个通道字段都有独立的初始化过程
通道 interrupt 被初始化为缓冲区容量为 1 的通道。这可以保证通道至少能接收一个来自语言运行时的 os.Signal 值确保语言运行时发送这个事件的时候不会被阻塞。如果 goroutine没有准备好接收这个值这个值就会被丢弃。例如如果用户反复敲 CtrlC 组合键程序只会在这个通道的缓冲区可用的时候接收事件其余的所有事件都会被丢弃。
通道 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 (errorsosos/signaltime
)// 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 (logostimecode/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
