共计 10978 个字符,预计需要花费 28 分钟才能阅读完成。
简介
之前写过一篇文章介绍了 ants
这个 goroutine 池实现。过后在网上查看相干材料的时候,发现了另外一个实现tunny
。趁着工夫相近,正好钻研一番。也好比拟一下这两个库。那就让咱们开始吧。
疾速开始
本文代码应用 Go Modules。
创立目录并初始化:
$ mkdir tunny && cd tunny
$ go mod init github.com/darjun/go-daily-lib/tunny
应用 go get
从 GitHub 获取 tunny
库:
$ go get -u github.com/Jeffail/tunny
为了不便地和 ants
做一个比照,咱们将 ants
中的示例从新用 tunny
实现一遍:还是那个分段求和的例子:
const (
DataSize = 10000
DataPerTask = 100
)
func main() {numCPUs := runtime.NumCPU()
p := tunny.NewFunc(numCPUs, func(payload interface{}) interface{} {
var sum int
for _, n := range payload.([]int) {sum += n}
return sum
})
defer p.Close()
// ...
}
应用也非常简单,首先创立一个Pool
,这里应用tunny.NewFunc()
。
第一个参数为池子大小,即同时有多少个 worker(也即 goroutine)在工作,这里设置成逻辑 CPU 个数,对于 CPU 密集型工作,这个值设置太大无意义,反而有可能导致 goroutine 切换频繁而升高性能。
第二个参数传入一个 func(interface{})interface{}
的参数作为工作处理函数。后续传入数据就会调用这个函数解决。
池子应用完须要敞开,这里应用 defer p.Close()
在程序退出前敞开。
而后,生成测试数据,还是 10000 个随机数,分成 100 组:
nums := make([]int, DataSize)
for i := range nums {nums[i] = rand.Intn(1000)
}
解决每组数据:
var wg sync.WaitGroup
wg.Add(DataSize / DataPerTask)
partialSums := make([]int, DataSize/DataPerTask)
for i := 0; i < DataSize/DataPerTask; i++ {go func(i int) {partialSums[i] = p.Process(nums[i*DataPerTask : (i+1)*DataPerTask]).(int)
wg.Done()}(i)
}
wg.Wait()
调用 p.Process()
办法,传入工作数据,池子中会抉择闲暇的 goroutine 来解决这个数据。因为咱们下面设置了处理函数,goroutine 会间接调用该函数,将这个切片作为参数传入。
tunny
与 ants
不同的是,tunny
的工作解决是同步的,即调用 p.Process()
办法之后,以后 goroutine 会挂起,直到工作解决实现之后才会被唤醒。因为是同步的,所以 p.Process()
办法能够间接返回处理结果。这也是下面程序在散发工作的时候,启动多个 goroutine 的起因。如果不是每个工作都启动一个 goroutine,p.Process()
办法会始终期待工作实现,那么前面的工作要等到后面的工作全副执行完之后能力执行。这样就施展不了并发的劣势了。
这里留神一个小细节,我将 for
循环变量作为参数传给 goroutine 函数了。如果不这样做,所有 goroutine 都共用外层的 i
,而且 goroutine 开始运行时,for
循环大概率曾经完结了,这时 i = DataSize/DataPerTask
,索引nums[i*DataPerTask : (i+1)*DataPerTask]
会越界触发 panic。
最初统计数据,验证后果:
var sum int
for _, s := range partialSums {sum += s}
var expect int
for _, num := range nums {expect += num}
fmt.Printf("finish all tasks, result is %d expect:%d\n", sum, expect)
运行:
$ go run main.go
finish all tasks, result is 5010172 expect:5010172
超时
默认状况下,p.Process()
会始终阻塞直到工作实现,即便以后没有闲暇 worker 也会阻塞。咱们也能够应用带超时的 Process()
办法:ProcessTimed()
。传入一个超时工夫距离,如果超过这个工夫还没有闲暇 worker,或者工作还没有解决实现,就会终止,并返回一个谬误。
超时有 2 种状况:
- 等不到闲暇的 worker:所有 worker 始终解决忙碌状态,正在解决的工作比拟耗时,无奈短时间内实现;
- 工作自身比拟耗时。
上面咱们编写一个计算斐波那契的函数,应用递归这种低效的实现办法:
func fib(n int) int {
if n <= 1 {return 1}
return fib(n-1) + fib(n-2)
}
咱们先看工作比拟耗时的状况,创立 Pool
对象。为了察看更显著,在处理函数中增加了 time.Sleep()
语句:
p := tunny.NewFunc(numCPUs, func(payload interface{}) interface{} {n := payload.(int)
result := fib(n)
time.Sleep(5 * time.Second)
return result
})
defer p.Close()
生成与池容量相等的工作数,调用 p.ProcessTimed()
办法,设置超时为 1s:
var wg sync.WaitGroup
wg.Add(numCPUs)
for i := 0; i < numCPUs; i++ {go func(i int) {n := rand.Intn(30)
result, err := p.ProcessTimed(n, time.Second)
nowStr := time.Now().Format("2006-01-02 15:04:05")
if err != nil {fmt.Printf("[%s]task(%d) failed:%v\n", nowStr, i, err)
} else {fmt.Printf("[%s]fib(%d) = %d\n", nowStr, n, result)
}
wg.Done()}(i)
}
wg.Wait()
因为处理函数中 sleep 5s,所以工作在执行过程中就超时了。运行:
$ go run main.go
[2021-06-10 16:36:26]task(7) failed:job request timed out
[2021-06-10 16:36:26]task(4) failed:job request timed out
[2021-06-10 16:36:26]task(1) failed:job request timed out
[2021-06-10 16:36:26]task(6) failed:job request timed out
[2021-06-10 16:36:26]task(5) failed:job request timed out
[2021-06-10 16:36:26]task(0) failed:job request timed out
[2021-06-10 16:36:26]task(3) failed:job request timed out
[2021-06-10 16:36:26]task(2) failed:job request timed out
都在同一秒中超时。
咱们将工作数量翻倍,再将处理函数中的 sleep 改为 990ms,保障前一批工作能顺利完成,后续工作或者因为等不到闲暇 worker,或者因为执行工夫过长而超时返回。运行:
$ go run main.go
[2021-06-10 16:42:46]fib(11) = 144
[2021-06-10 16:42:46]fib(25) = 121393
[2021-06-10 16:42:46]fib(27) = 317811
[2021-06-10 16:42:46]fib(1) = 1
[2021-06-10 16:42:46]fib(18) = 4181
[2021-06-10 16:42:46]fib(29) = 832040
[2021-06-10 16:42:46]fib(17) = 2584
[2021-06-10 16:42:46]fib(20) = 10946
[2021-06-10 16:42:46]task(5) failed:job request timed out
[2021-06-10 16:42:46]task(14) failed:job request timed out
[2021-06-10 16:42:46]task(8) failed:job request timed out
[2021-06-10 16:42:46]task(7) failed:job request timed out
[2021-06-10 16:42:46]task(13) failed:job request timed out
[2021-06-10 16:42:46]task(12) failed:job request timed out
[2021-06-10 16:42:46]task(11) failed:job request timed out
[2021-06-10 16:42:46]task(6) failed:job request timed out
context
context 是协调 goroutine 的工具。tunny
反对带 context.Context
参数的办法:ProcessCtx()
。以后 context 状态变为 Done
之后,工作也会进行执行。context 会因为超时、勾销等起因切换为 Done
状态。还是拿下面的例子:
go func(i int) {n := rand.Intn(30)
ctx, cancel := context.WithCancel(context.Background())
if i%2 == 0 {go func() {time.Sleep(500 * time.Millisecond)
cancel()}()}
result, err := p.ProcessCtx(ctx, n)
if err != nil {fmt.Printf("task(%d) failed:%v\n", i, err)
} else {fmt.Printf("fib(%d) = %d\n", n, result)
}
wg.Done()}(i)
其余代码都一样,咱们调用 p.ProcessCtx()
办法来执行工作。参数是一个可勾销的 Context
。对于序号为偶数的工作,咱们启动一个 goroutine 在 500ms 之后cancel()
掉这个Context
。代码运行后果如下:
$ go run main.go
task(4) failed:context canceled
task(6) failed:context canceled
task(0) failed:context canceled
task(2) failed:context canceled
fib(27) = 317811
fib(25) = 121393
fib(1) = 1
fib(18) = 4181
咱们看到偶数序号的工作都被勾销了。
源码
tunny
的源码更少,除去测试代码和正文,连 500 行都不到。那么就一起来看一下吧。Pool
构造如下:
// src/github.com/Jeffail/tunny.go
type Pool struct {
queuedJobs int64
ctor func() Worker
workers []*workerWrapper
reqChan chan workRequest
workerMut sync.Mutex
}
Pool
构造中有一个 ctor
字段,这是一个函数对象,用于返回一个实现 Worker
接口的值:
type Worker interface {Process(interface{}) interface{}
BlockUntilReady()
Interrupt()
Terminate()}
这个接口不同的办法在工作执行的不同阶段调用。最重要的当属 Process(interface{}) interface{}
办法了。这个就是执行工作的函数。tunny
提供 New()
办法创立 Pool
对象,这个办法须要咱们本人结构 ctor
函数对象,应用多有不便。tunny
提供了另外两个默认实现 closureWorker
和callbackWorker
:
type closureWorker struct {processor func(interface{}) interface{}}
func (w *closureWorker) Process(payload interface{}) interface{} {return w.processor(payload)
}
func (w *closureWorker) BlockUntilReady() {}
func (w *closureWorker) Interrupt() {}
func (w *closureWorker) Terminate() {}
type callbackWorker struct{}
func (w *callbackWorker) Process(payload interface{}) interface{} {f, ok := payload.(func())
if !ok {return ErrJobNotFunc}
f()
return nil
}
func (w *callbackWorker) BlockUntilReady() {}
func (w *callbackWorker) Interrupt() {}
func (w *callbackWorker) Terminate() {}
tunny.NewFunc()
办法应用的就是closureWorker
:
func NewFunc(n int, f func(interface{}) interface{}) *Pool {return New(n, func() Worker {
return &closureWorker{processor: f,}
})
}
创立的 closureWorker
间接将参数 f
作为工作处理函数。
tunny.NewCallback()
办法应用callbackWorker
:
func NewCallback(n int) *Pool {return New(n, func() Worker {return &callbackWorker{}
})
}
callbackWorker
构造中没有处理函数,只能给它发送无参无返回值的函数对象作为工作,它的 Process()
办法就是执行这个函数。
创立 Pool
对象后,都是调用它的 SetSize()
办法,设置 worker 数量。在这个办法中会启动相应数量的 goroutine:
func (p *Pool) SetSize(n int) {p.workerMut.Lock()
defer p.workerMut.Unlock()
lWorkers := len(p.workers)
if lWorkers == n {return}
for i := lWorkers; i < n; i++ {p.workers = append(p.workers, newWorkerWrapper(p.reqChan, p.ctor()))
}
// 进行过多的 worker
for i := n; i < lWorkers; i++ {p.workers[i].stop()}
// 期待 worker 进行
for i := n; i < lWorkers; i++ {p.workers[i].join()
// -----------------
}
p.workers = p.workers[:n]
}
SetSize()
其实在扩容和缩容的时候也会调用。对于扩容,它会创立相应数量的 worker。对于缩容,它会将多余的 worker 停掉。与 ants
不同,tunny
的扩容缩容都是即时失效的。
代码中,我用 -----------------
标出来的中央我感觉有点问题。对于缩容,因为底层的数组没有变动,workers
切片长度放大之后,数组中前面的元素实际上就拜访不到了,然而数组还持有它的援用,算是一种内存透露吧。所以稳当起见最好加上p.workers[i] = nil
?
这里创立的 worker 实际上是包装了一层的 workerWrapper
构造:
// src/github.com/Jeffail/worker.go
type workerWrapper struct {
worker Worker
interruptChan chan struct{}
reqChan chan<- workRequest
closeChan chan struct{}
closedChan chan struct{}}
func newWorkerWrapper(
reqChan chan<- workRequest,
worker Worker,
) *workerWrapper {
w := workerWrapper{
worker: worker,
interruptChan: make(chan struct{}),
reqChan: reqChan,
closeChan: make(chan struct{}),
closedChan: make(chan struct{}),
}
go w.run()
return &w
}
workerWrapper
构造创立之后会立即调用 run()
办法启动一个 goroutine:
func (w *workerWrapper) run() {jobChan, retChan := make(chan interface{}), make(chan interface{})
defer func() {w.worker.Terminate()
close(retChan)
close(w.closedChan)
}()
for {w.worker.BlockUntilReady()
select {
case w.reqChan <- workRequest{
jobChan: jobChan,
retChan: retChan,
interruptFunc: w.interrupt,
}:
select {
case payload := <-jobChan:
result := w.worker.Process(payload)
select {
case retChan <- result:
case <-w.interruptChan:
w.interruptChan = make(chan struct{})
}
case _, _ = <-w.interruptChan:
w.interruptChan = make(chan struct{})
}
case <-w.closeChan:
return
}
}
}
每个 worker goroutine 都在尝试向 w.reqChan
通道中发送一个 workRequest
构造数据,发送胜利之后,从 jobChan
中获取工作数据,而后调用 Worker.Process()
办法执行工作,最初将后果发送到 retChan
通道中。这里其实有好几个交互。须要联合 Process()
办法来看才更清晰:
func (p *Pool) Process(payload interface{}) interface{} {
request, open := <-p.reqChan
request.jobChan <- payload
payload, open = <-request.retChan
return payload
}
删掉无相干的代码,最初就是下面这样。咱们在调用池对象的 Process()
办法时,尝试从通道 reqChan
中接收数据,而后将工作数据发送到 jobChan
通道中,最初从 retChan
通道中接管后果。与下面的 run
流程联合来看,实际上在失常执行一个工作时,Pool
与 workerWrapper
有 3 次交互。
察看 Pool
创立到 workerWrapper
创立的流程,咱们能够看到实际上 Pool
构造中的 reqChan
与workerWrapper
构造中的 reqChan
是同一个通道。即 workerWrapper
启动后,会阻塞在向 reqChan
通道发送数据上,直到调用了 Pool
的Process*()
办法,从通道 reqChan
取出数据。Process()
办法失去 workRequest
会向它的 jobChan
通道中发送工作数据。而 workerWrapper.run()
办法胜利发送数据到 reqChan
之后就开始期待从 jobChan
通道中接收数据,这时接管到 Process()
办法发送过去的数据。开始执行 w.worker.Process()
办法,而后向 retChan
通道发送后果数据,Process()
办法在胜利发送数据到 jobChan
之后,就开始期待从 retChan
通道中接收数据。接管胜利之后,Process()
办法返回,workerWrapper.run()
持续阻塞在 w.reqChan <-
这条语句上,期待解决下一个工作。留神 jobChan
和retChan
都是 workerWrapper.run()
办法中创立的通道。
那么超时是怎么实现的呢?看办法 ProcessTimed()
的实现:
func (p *Pool) ProcessTimed(payload interface{},
timeout time.Duration,
) (interface{}, error) {tout := time.NewTimer(timeout)
var request workRequest
select {
case request, open = <-p.reqChan:
case <-tout.C:
return nil, ErrJobTimedOut
}
select {
case request.jobChan <- payload:
case <-tout.C:
request.interruptFunc()
return nil, ErrJobTimedOut
}
select {
case payload, open = <-request.retChan:
case <-tout.C:
request.interruptFunc()
return nil, ErrJobTimedOut
}
tout.Stop()
return payload, nil
}
同样地,删除不相干的代码。首先,创立一个 timer
,超时工夫由传入参数指定。前面有 3 个select
语句:
- 期待从
p.reqChan
取数据,即期待有 worker 闲暇; - 期待发送数据到
jobChan
,即期待 worker 从jobChan
取出工作数据; - 期待从
retChan
取数据,即期待 worker 将后果发送到retChan
。
第一种状况,如果超时了,阐明 worker 都处于忙碌状态,间接返回工作超时。前面两种状况实际上是工作曾经开始执行了,然而在规定的工夫内没有实现。这两种状况,须要终止工作的执行。咱们看到下面调用了 workerRequest.interruptFunc()
办法,也就是 workerWrapper.interrupt()
办法:
func (w *workerWrapper) interrupt() {close(w.interruptChan)
w.worker.Interrupt()}
这个办法就是简略敞开了 interrupteChan
通道,而后调用 worker
对象的 Interrupt()
办法,默认实现中这个办法都是空的。
interruptChan
通道敞开后,goroutine 中期待从 jobChan
接收数据和期待向 retChan
发送数据的操作都会勾销:
select {
case payload := <-jobChan:
result := w.worker.Process(payload)
select {
case retChan <- result:
case <-w.interruptChan:
w.interruptChan = make(chan struct{})
}
case _, _ = <-w.interruptChan:
w.interruptChan = make(chan struct{})
}
ProcessCtx()
实现也是相似的。
最初调用 workerWrapper.stop()
会敞开 closeChan
通道,这会导致 workerWrapper.run()
办法中的 for
循环跳出,进而执行 defer
函数中的 close(retChan)
和close(closedChan)
:
defer func() {w.worker.Terminate()
close(retChan)
close(w.closedChan)
}()
这里须要敞开 retChan
通道是为了避免 Process*()
办法在期待 retChan
数据。
closedChan
通道敞开后,workerWrapper.join()
办法就返回了。
func (w *workerWrapper) join() {<-w.closedChan}
Worker
几个办法的调用机会:
Process()
:执行工作时;Interrupt()
:工作因为超时会被 context 勾销时;BlockUntilReady()
:每次执行新工作前,可能须要筹备一些资源;Terminate()
:workerWrapper.run()
中的 defer 函数中,即进行 worker 后。
这些机会在代码中都能清晰地看到。
基于源码,我画了一个流程图:
图中省略了中断的流程。
tunny
vs ants
tunny
设计的思路与 ants
有较大的区别:
tunny
只反对同步的形式执行工作,尽管工作在另一个 goroutine 执行,然而提交工作的 goroutine 必须期待后果返回或超时。不能做其余事件。正是因为这一点,导致 tunny
的设计略微一点简单,而且为了反对超时和勾销,设计了多个通道用于和执行工作的 goroutine 通信。一次工作执行的过程波及屡次通信,性能是有损失的。从另一方面说,同步的编程形式更合乎人类的直觉。
ants
齐全是异步的工作执行流程,相比 tunny
性能是稍高一些的。然而也因为它的异步个性,导致没有工作超时、勾销这些机制。而且如果须要收集后果,必须要本人编写额定的代码。
总结
本文介绍了另一个 goroutine 池的实现 tunny
。它以同步的形式来解决工作,编写代码更加直观,对工作的执行流程有更强的管制,如超时、勾销等。当然实现也简单一些。tunny
代码不走 500 行,十分倡议读一读。
大家如果发现好玩、好用的 Go 语言库,欢送到 Go 每日一库 GitHub 上提交 issue😄
参考
- tunny GitHub:https://github.com/Jeffail/tunny
- ants GitHub:github.com/panjf2000/ants
- Go 每日一库 GitHub:https://github.com/darjun/go-daily-lib
我
我的博客:https://darjun.github.io
欢送关注我的微信公众号【GoUpUp】,独特学习,一起提高~