【并发模式】Go 常见并发模式实现Runner、Pool、Work

news2024/11/22 18:17:23

通过并发编程在 Go 程序中实现的3种常见的并发模式。

参考：https://cloud.tencent.com/developer/article/1720733

1、Runner 定时任务

Runner 模式有代表性，能把（任务队列，超时，系统中断信号）等结合起来形成一项定时任务。任何一个条件满足触发，程序就结束了。

import (
 "errors"
 "os"
 "os/signal"
 "time"
)

//Runner 在给定的超时时间内执行一组任务
// 并且在操作系统发送中断信号时结束这些任务
type Runner struct {
 //从操作系统发送信号
 interrupt chan os.Signal
 //报告处理任务已完成
 complete chan error
 //报告处理任务已经超时
 timeout <-chan time.Time
 //持有一组以索引为顺序的依次执行的以 int 类型 id 为参数的函数
 tasks []func(id int)
}

//New 函数返回一个新的准备使用的 Runner，d：自定义分配的时间
func New(d time.Duration) *Runner {
 return &Runner{
  interrupt: make(chan os.Signal, 1),
  complete:  make(chan error),
  //会在另一线程经过时间段 d 后向返回值发送当时的时间。
  timeout: time.After(d),
 }
}

//Add 将一个任务加入到 Runner 中
func (r *Runner) Add(tasks ...func(id int)) {

}

//Start 开始执行所有任务，并监控通道事件
func (r *Runner) Start() error {

}

//执行每一个已注册的任务
func (r *Runner) run() error {
}

//检测是否收到了中断信号
func (r *Runner) gotInterrupt() bool {

}

Runner 类型声明了 3 个通道，用来辅助管理程序的生命周期，以及用来表示顺序执行的不同任务的函数切片。

2、Pool 缓存池

Go 1.6 及之后的版本中，标准库里自带有资源池的实现：sync.Pool，（还有更好的gopool，后者能控制协程数。）

sync.Pool: New、Get、Put

//
func main() {
	// 创建一个 sync.Pool
	var pool sync.Pool

	// 设置对象池的 New 函数，创建对象
	pool.New = func() interface{} {
		return "New Object"
	}

	// 获取对象
	obj := pool.Get()
	fmt.Println(obj) // 输出：New Object

	// 将对象放回池中
	pool.Put("Reused Object")

	// 再次获取对象
	obj = pool.Get()
	fmt.Println(obj) // 输出：Reused Object
}

func main() {
	var studentPool = sync.Pool{
        New: func() interface{} { 
            return new(Student) // 例如创建 Student 对象
        },
    }
    
    //取得对象和归还对象
    stu := studentPool.Get().(*Student)
    json.Unmarshal(buf, stu) // 使用线程池的对象
    studentPool.Put(stu)
    
    //Get() 用于从对象池中获取对象，因为返回值是 interface{}，因此需要类型转换。Put() 则是在对象使用完毕后，返回对象池。
 }

gopool ： NewPool、CtxGo

高性能：针对高并发进行了优化，性能较 sync.Pool 更高。
灵活性强：支持更多的池管理选项，比如对象的最大数量、超时时间等。

pool := gopool.NewPool("article_spider", 10, gopool.NewConfig())
//for {}
pool.CtxGo(ctx, func() {
				defer wg.Done()

				c.dbMutex.Lock()
				defer c.dbMutex.Unlock()

				//处理程序
			})