文章详情

短信预约-IT技能 免费直播动态提醒

请输入下面的图形验证码

提交验证

短信预约提醒成功

Golang协程池gopool怎么设计与实现

2023-06-30 03:39

关注

这篇文章主要介绍“Golang协程池gopool怎么设计与实现”,在日常操作中,相信很多人在Golang协程池gopool怎么设计与实现问题上存在疑惑,小编查阅了各式资料,整理出简单好用的操作方法,希望对大家解答”Golang协程池gopool怎么设计与实现”的疑惑有所帮助!接下来,请跟着小编一起来学习吧!

Goroutine

Goroutine 是 Golang 提供的一种轻量级线程,我们通常称之为「协程」,相比较线程,创建一个协程的成本是很低的。所以你会经常看到 Golang 开发的应用出现上千个协程并发的场景。

Goroutine 的优势:

它们的堆栈大小只有几 kb,堆栈可以根据应用程序的需要增长和缩小,context switch 也很快,而在线程的情况下,堆栈大小必须指定并固定。

一个包含数千个 Goroutine 的程序中可能只有一个线程。如果该线程中的任何 Goroutine 阻塞等待用户输入,则创建另一个 OS 线程并将剩余的 Goroutine 移动到新的 OS 线程。所有这些都由运行时处理,作为开发者无需耗费心力关心,这也使得我们有很干净的 API 来支持并发。

channel 的设计有效防止了在使用 Goroutine 访问共享内存时发生竞争条件(race conditions) 。channel 可以被认为是 Goroutine 进行通信的管道。

协程池

在高并发场景下,我们可能会启动大量的协程来处理业务逻辑。协程池是一种利用池化技术,复用对象,减少内存分配的频率以及协程创建开销,从而提高协程执行效率的技术。

最近抽空了解了字节官方开源的 gopkg 库提供的 gopool 协程池实现,感觉还是很高质量的,代码也非常简洁清晰,而且 Kitex 底层也在使用 gopool 来管理协程,这里我们梳理一下设计和实现。

gopool

了解官方 README 就会发现gopool的用法其实非常简单,将曾经我们经常使用的 go func(){...} 替换为 gopool.Go(func(){...}) 即可。

此时 gopool 将会使用默认的配置来管理你启动的协程,你也可以选择针对业务场景配置池子大小,以及扩容上限。

old:

go func() {// do your job}()

new:

import (    "github.com/bytedance/gopkg/util/gopool")gopool.Go(func(){/// do your job})

核心实现

下面我们来看看gopool是怎样实现协程池管理的。

Pool

Pool 是一个定义了协程池能力的接口。

type Pool interface {// 池子的名称Name() string        // 设置池子内Goroutine的容量SetCap(cap int32)        // 执行 f 函数Go(f func())        // 带 ctx,执行 f 函数CtxGo(ctx context.Context, f func())        // 设置发生panic时调用的函数SetPanicHandler(f func(context.Context, interface{}))}

gopool 提供了这个接口的默认实现(即下面即将介绍的pool),当我们直接调用 gopool.CtxGo 时依赖的就是这个。

这样的设计模式在 Kitex 中也经常出现,所有的依赖均设计为接口,便于随后扩展,底层提供一个默认的实现暴露出去,这样对调用方也很友好。

type pool struct {// 池子名称name string// 池子的容量, 即最大并发工作的 goroutine 的数量cap int32        // 池子配置config *Config        // task 链表taskHead  *tasktaskTail  *tasktaskLock  sync.MutextaskCount int32// 记录当前正在运行的 worker 的数量workerCount int32// 当 worker 出现panic时被调用panicHandler func(context.Context, interface{})}// NewPool 创建一个新的协程池,初始化名称,容量,配置func NewPool(name string, cap int32, config *Config) Pool {p := &pool{name:   name,cap:    cap,config: config,}return p}

调用 NewPool 获取了以 Pool 的形式返回的 pool 结构体。

Task

type task struct {ctx context.Contextf   func()next *task}

task 是一个链表结构,可以把它理解为一个待执行的任务,它包含了当前节点需要执行的函数f, 以及指向下一个task的指针。

综合前一节 pool 的定义,我们可以看到,一个协程池 pool 对应了一组task

pool 维护了指向链表的头尾的两个指针:taskHeadtaskTail,以及链表的长度taskCount 和对应的锁 taskLock

Worker

type worker struct {pool *pool}

一个 worker 就是逻辑上的一个执行器,它唯一对应到一个协程池 pool。当一个worker被唤起,将会开启一个goroutine ,不断地从 pool 中的 task链表获取任务并执行。

func (w *worker) run() {go func() {for {                        // 声明即将执行的 taskvar t *task                                                // 操作 pool 中的 task 链表,加锁w.pool.taskLock.Lock()if w.pool.taskHead != nil {                                // 拿到 taskHead 准备执行t = w.pool.taskHead                                                                // 更新链表的 head 以及数量w.pool.taskHead = w.pool.taskHead.nextatomic.AddInt32(&w.pool.taskCount, -1)}                        // 如果前一步拿到的 taskHead 为空,说明无任务需要执行,清理后返回if t == nil {w.close()w.pool.taskLock.Unlock()w.Recycle()return}w.pool.taskLock.Unlock()                                                // 执行任务,针对 panic 会recover,并调用配置的 handlerfunc() {defer func() {if r := recover(); r != nil {msg := fmt.Sprintf("GOPOOL: panic in pool: %s: %v: %s", w.pool.name, r, debug.Stack())logger.CtxErrorf(t.ctx, msg)if w.pool.panicHandler != nil {w.pool.panicHandler(t.ctx, r)}}}()t.f()}()t.Recycle()}}()}

整体来看

看到这里,其实就能把整个流程串起来了。我们来看看对外的接口 CtxGo(context.Context, f func()) 到底做了什么?

func Go(f func()) {CtxGo(context.Background(), f)}func CtxGo(ctx context.Context, f func()) {defaultPool.CtxGo(ctx, f)}func (p *pool) CtxGo(ctx context.Context, f func()) {        // 创建一个 task 对象,将 ctx 和待执行的函数赋值t := taskPool.Get().(*task)t.ctx = ctxt.f = f                // 将 task 插入 pool 的链表的尾部,更新链表数量p.taskLock.Lock()if p.taskHead == nil {p.taskHead = tp.taskTail = t} else {p.taskTail.next = tp.taskTail = t}p.taskLock.Unlock()atomic.AddInt32(&p.taskCount, 1)                // 以下两个条件满足时,创建新的 worker 并唤起执行:// 1. task的数量超过了配置的限制 // 2. 当前运行的worker数量小于上限(或无worker运行)if (atomic.LoadInt32(&p.taskCount) >= p.config.ScaleThreshold && p.WorkerCount() < atomic.LoadInt32(&p.cap)) || p.WorkerCount() == 0 {                        // worker数量+1p.incWorkerCount()                                // 创建一个新的worker,并把当前 pool 赋值w := workerPool.Get().(*worker)w.pool = p                                // 唤起worker执行w.run()}}

相信看了代码注释,大家就能理解发生了什么。

gopool 会自行维护一个 defaultPool,这是一个默认的 pool 结构体,在引入包的时候就进行初始化。当我们直接调用 gopool.CtxGo() 时,本质上是调用了 defaultPool 的同名方法

func init() {defaultPool = NewPool("gopool.DefaultPool", 10000, NewConfig())}const (defaultScalaThreshold = 1)// Config is used to config pool.type Config struct {// 控制扩容的门槛,一旦待执行的 task 超过此值,且 worker 数量未达到上限,就开始启动新的 workerScaleThreshold int32}// NewConfig creates a default Config.func NewConfig() *Config {c := &Config{ScaleThreshold: defaultScalaThreshold,}return c}

defaultPool 的名称为 gopool.DefaultPool,池子容量一万,扩容下限为 1。

当我们调用 CtxGo时,gopool 就会更新维护的任务链表,并且判断是否需要扩容 worker

三个角色的定位

使用 sync.Pool 进行性能优化

其实到这个地方,gopool已经是一个代码简洁清晰的协程池库了,但是性能上显然有改进空间,所以gopool的作者应用了多次 sync.Pool 来池化对象的创建,复用woker和task对象。

这里建议大家直接看源码,其实在上面的代码中已经有所涉及。

var taskPool sync.Poolfunc init() {taskPool.New = newTask}func newTask() interface{} {return &task{}}func (t *task) Recycle() {t.zero()taskPool.Put(t)}
var workerPool sync.Poolfunc init() {workerPool.New = newWorker}func newWorker() interface{} {return &worker{}}func (w *worker) Recycle() {w.zero()workerPool.Put(w)}

到此,关于“Golang协程池gopool怎么设计与实现”的学习就结束了,希望能够解决大家的疑惑。理论与实践的搭配能更好的帮助大家学习,快去试试吧!若想继续学习更多相关知识,请继续关注编程网网站,小编会继续努力为大家带来更多实用的文章!

阅读原文内容投诉

免责声明:

① 本站未注明“稿件来源”的信息均来自网络整理。其文字、图片和音视频稿件的所属权归原作者所有。本站收集整理出于非商业性的教育和科研之目的,并不意味着本站赞同其观点或证实其内容的真实性。仅作为临时的测试数据,供内部测试之用。本站并未授权任何人以任何方式主动获取本站任何信息。

② 本站未注明“稿件来源”的临时测试数据将在测试完成后最终做删除处理。有问题或投稿请发送至: 邮箱/279061341@qq.com QQ/279061341

软考中级精品资料免费领

  • 历年真题答案解析
  • 备考技巧名师总结
  • 高频考点精准押题
  • 2024年上半年信息系统项目管理师第二批次真题及答案解析(完整版)

    难度     813人已做
    查看
  • 【考后总结】2024年5月26日信息系统项目管理师第2批次考情分析

    难度     354人已做
    查看
  • 【考后总结】2024年5月25日信息系统项目管理师第1批次考情分析

    难度     318人已做
    查看
  • 2024年上半年软考高项第一、二批次真题考点汇总(完整版)

    难度     435人已做
    查看
  • 2024年上半年系统架构设计师考试综合知识真题

    难度     224人已做
    查看

相关文章

发现更多好内容

猜你喜欢

AI推送时光机
位置:首页-资讯-后端开发
咦!没有更多了?去看看其它编程学习网 内容吧
首页课程
资料下载
问答资讯