go并发设计模式runner模式

🕗 发布于 2024-12-02 10:46 golang 设计模式

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() string
    Signal() // 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 uint64
ext  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 {
// 如果有中断信号那么返回true
case <-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 runner

import (
"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 {
// 如果有中断信号那么返回true
case <-r.interrupt:
// 接收到中断信号，停止后续再接收到中断信号
signal.Stop(r.interrupt)
return true
// 没有终端信号返回false，继续执行
default:
return false
}
}

在main.go中进行调用

package main

import (
"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

原文地址：https://blog.csdn.net/andrewgithub/article/details/144168415

免责声明：本站文章内容转载自网络资源，如本站内容侵犯了原著者的合法权益，可联系本站删除。更多内容请关注自学内容网（zxcms.com）！

上一篇：Hive分区表新增字段并指定位置
下一篇：文件比较和文件流

构建专属AI知识库：Obsidian Copilot + 硅基流动SiliconCloud API 实战指南
在信息爆炸的时代，如何有效地管理和利用个人或团队的知识资产成为了一个关键问题。对于那些寻求提高效率和创新能力的人来说，拥有一个智能化的知识管理系统显得尤为重要。本文将向您介绍如何结合Obsidian
阅读更多2024-12-26
使用GPT进行SCI论文润色常用语句
转载自：【实例演示ChatGPT论文润色，精准降重和翻译（毕业写作必看）】https://www.bilibili.com/video/BV1z142197Nm?声明：本文仅作为本人记录学习使用。
阅读更多2024-12-26
科普时刻 | 借助鲁棒性设计解决制造中的不确定性问题
实际上，在现场测试中其实无法达到评估系统故障概率所需的里程数，因此，汽车制造商梅赛德斯-奔驰公司转而利用仿真来确定关键交通场景，并使用optiSLang进行安全性功能测试和评估。为了实现准确的公差分析
阅读更多2024-12-26
wtforms+flask_sqlalchemy在flask-admin视图下实现日期的修改与更新
在flask-admin 的modelview视图下实现自定义视图的表单修改/编辑是件不太那么容易的事情，特别是想不自定义前端view的情况下。
阅读更多2024-12-26
【蓝桥杯每日一题】分糖果——DFS
蓝桥杯每日一题 2024-12-24 分糖果 DFS
阅读更多2024-12-26
使用ForceBindIP绑定应用到指定IP
使用ForceBindIP工具，用户可以轻松地将特定应用程序绑定到指定的IP地址，从而确保应用程序的网络连接通过指定的网络适配器进行。通过在命令提示符下运行ForceBindIP并指定IP地址和应用程
阅读更多2024-12-26
纯真社区版IP库CZDB数据格式使用教程
纯真社区版IP库是一种免费且公开的IP地理位置数据库，主要面向非商业用途。其最新推出的CZDB格式是一种全新的数据文件格式，自2024年10月起将成为官方维护和更新的唯一版本。
阅读更多2024-12-26
CSharp: Oracle Stored Procedure query table
【代码】CSharp: Oracle Stored Procedure query table。
阅读更多2024-12-26
Spring02 - 代理和事务篇
Spring02 - 代理和事务篇
阅读更多2024-12-26
Spring Boot使用多线程
使用配置线程池。使用@Async注解来异步执行方法。可以自定义线程池的配置（如核心线程数、最大线程数等）。通过或Future获取异步任务的执行结果。这种方式不仅能提高程序的并发性能，还能更好地管理线程
阅读更多2024-12-26

go并发设计模式runner模式