Mix XWP V1.1 - Go 通用动态协程池 WorkerPool

[onanying]

OpenMix 出品：https://openmix.org

Mix XWP

通用动态工作池、协程池

A common worker pool

Github

https://github.com/mix-go/xwp

Installation

go get github.com/mix-go/xwp

Usage

先创建一个结构体用来处理任务，使用类型断言转换任务数据类型，例如：i := data.(int)

type Foo struct {
}

func (t *Foo) Do(data interface{}) {
    // do something
}

调度任务

• 也可以使用 RunF 采用闭包来处理任务
• 如果不想阻塞执行，可以使用 p.Start() 启动

jobQueue := make(chan interface{}, 200)

p := &xwp.WorkerPool{
    JobQueue:       jobQueue,
    MaxWorkers:     1000,
    InitWorkers:    100,
    MaxIdleWorkers: 100,
    RunI:           &Foo{},
}

go func() {
    // 投放任务
    for i := 0; i < 10000; i++ {
        jobQueue <- i
    }

    // 投放完停止调度
    p.Stop()
}()

p.Run() // 阻塞等待

异常处理：Do 方法中执行的代码，可能会出现 panic 异常，我们可以通过 recover 获取异常信息记录到日志或者执行其他处理

func (t *Foo) Do(data interface{}) {
    defer func() {
        if err := recover(); err != nil {
            // handle error
        }
    }()
    // do something
}

查看 Workers 的执行状态：通常可以使用一个定时器，定时打印或者告警处理

go func() {
    ticker := time.NewTicker(1000 * time.Millisecond)
    for {
        <-ticker.C
        log.Printf("%+v", p.Stat()) // 2021/04/26 14:32:53 &{Active:5 Idle:95 Total:100}
    }
}()

License

Apache License Version 2.0, http://www.apache.org/licenses/

[Mohanson]

池化解决的是资源频繁创建和销毁的开销

再想想协程本身就是轻量级线程, 不走内核调度的, 创建和销毁开销很小

你把协程池化的意义是什么

要多想

[pabupa]

@Mohanson 好像是为了避免调用 runtime.shrinkstack 。https://github.com/grpc/grpc-go/pull/3204
不过我也觉得没有必要。

[MarlonFan]

@Mohanson 极致性能场景, 大量协程情况下 Golang 自身的调度切换也是不小的压力, 所以尽可能的避免不断新创建协程. 参见 ``gnet`` , ``fasthttp``之类的实现

[pabupa]

@MarlonFan 如果真有那么极致的场景，想必业务一定很大，至少不缺加机器的钱吧……

[zeromake]

@Mohanson
我在写 go 实际情况下协程池还是有些用的，主要是高频访问产生的协程，但是协程的消费比不上生成，而且 go 在协程数量过多时调度协程也是一个蛋疼的情况。

[zeromake]

@pabupa
是不缺但是不加协程池访问量一旦暴增就会直接把服务跑挂。

[MarlonFan]

@zeromake +1

[kxuanobj]

有性能测试、对比吗？用这个和不用这个有什么性能上的区别？

[sujin190]

@zeromake #5 但这个分明是协程创建后实际执行时间和创建销毁调度协程所用时间比值太小问题，协程池解决这个也挺鸡肋的，既然如此就应该好好考虑短时任务长时化或者批量化才对吧

[GoLand]

@zeromake 访问量激增导致打挂服务和 goroutine 多少没关系，goroutine 这么轻量的资源都分配不出来了，服务底层的数据库或者其他组件更不一定能抗住。资源分配不合理，限流做的不到位。

goroutine 池化本身就是脱裤子放屁。

[kksco]

@GoLand 也不能怎么说，把 goroutine 限制住还是有好处的，毕竟 0 runtime 的调度还在进化，1 gc 这种压力也大。。。

[lewinlan]

fasthttp 我看就是反面典型
访问量激增把服务跑挂那是设计问题，思路错了

[onanying]

@Mohanson @pabupa 并不是应用在 web api 这种场景的，计算场景中有很多，比如：下单能支持 5000 qps 这是异步入到 mq，但是 mq 的消费用 cli 消费，这个时候消费的 cli 程序里要调用 db，而 db 能支持 500 qps，当不限制 go 数量时，mq 的请求全部取出，变成了一个不断增长的 goroutine，而限制 goroutine 数量的 pool 在消费不过来的时候会把 chan 阻塞，进而会停止从 mq 取新的数据，api 如果做了判断 mq size 的逻辑，接口就会返回错误让用户不再下单，但是没有用 pool 就会把 mq 的新数据不断的变成 new goroutine，直到内存无法分配。

[onanying]

当然上面说了：资源分配不合理，限流做的不到位 ，但是现实情况中如果代码健壮一些，是不是就不用这么去甩锅给做限流了呢？

[onanying]

还有并不是每个公司都是一台 8 cpu 的服务器只跑一个业务的，通常会跑很多种业务，用 pool 动态释放的内存，是有价值的，是可以让出给其他业务使用的

[Mohanson]

@onanying XY 问题. 你提的"使用场景", 正解是使用的是 ratelimit(令牌桶算法) 而不是一个 go 池.

[onanying]

哦，最大的意义复用协程我没有提，这个是 pool 最大的意义，和 db pool 复用连接一样，复用协程同样能节约 cpu，看着可能很小，但是任务一多起来，加起来差异就很大了。

[onanying]

@Mohanson 你可以这样认为，但是我觉得我做的项目我加了池，限流调大了也不会出问题，但是不加池，限流没做好久会挂，我觉得加上就更好，同样有些情况是突发的，令牌桶是可以续水的，如果桶的容量大，峰值超过了 db 的极限，也会出现上面我说的情况，但是实际不可能把桶设置这么小的，非常浪费资源。

[onanying]

其实还有一点，当 mq 有数据就直接 new goroutine，goroutine 数量很多，那 cli 程序重启就需要考虑必须要把正在执行的全部的 goroutine 全部执行完成才能停止进程，不然就会导致数据不一致的问题（ goroutine 处理一半，进程就 exit 了），用我写的这个 pool 就不许要考虑这个问题，因为我都封装好了。