worker pool其实就是线程池thread pool。对于go来说，直接使用的是goroutine而非线程，不过这里仍然以线程来解释线程池。
在线程池模型中，有2个队列一个池子：任务队列、已完成任务队列和线程池。其中已完成任务队列可能存在也可能不存在，依据实际需求而定。
只要有任务进来，就会放进任务队列中。只要线程执行完了一个任务，就将任务放进已完成任务队列，有时候还会将任务的处理结果也放进已完成队列中。
worker pool中包含了一堆的线程(worker，对go而言每个worker就是一个goroutine)，这些线程嗷嗷待哺，等待着为它们分配任务，或者自己去任务队列中取任务。取得任务后更新任务队列，然后执行任务，并将执行完成的任务放进已完成队列。
下图来自wiki：
在go中有两种方式可以实现工作池：传统的互斥锁、channel。
假设go中的任务的定义形式为：
我要评论type task struct {
...
}
每次有任务进来时，都将任务放在任务队列中。
使用传统的互斥锁方式实现，任务队列的定义结构大概如下：
type queue struct{
m sync.mutex
tasks []task
}
然后在执行任务的函数中加上lock()和unlock()。例如：
func worker(queue *queue) {
for {
// lock()和unlock()之间的是critical section
queue.m.lock()
// 取出任务
task := queue.tasks[0]
// 更新任务队列
queue.tasks = queue.tasks[1:]
queue.m.unlock()
// 在此goroutine中执行任务
process(task)
}
}
假如在线程池中激活了100个goroutine来执行worker()。lock()和unlock()保证了在同一时间点只能有一个goroutine取得任务并随之更新任务列表，取任务和更新任务队列都是critical section中的代码，它们是具有原子性。然后这个goroutine可以执行自己取得的任务。于此同时，其它goroutine可以争夺互斥锁，只要争抢到互斥锁，就可以取得任务并更新任务列表。当某个goroutine执行完process(task)，它将因为for循环再次参与互斥锁的争抢。
上面只是给出了一点主要的代码段，要实现完整的线程池，还有很多额外的代码。
通过互斥锁，上面的一切操作都是线程安全的。但问题在于加锁/解锁的机制比较重量级，当worker(即goroutine)的数量足够多，锁机制的实现将出现瓶颈。
在go中，也能用buffered channel实现工作池。
示例代码很长，所以这里先拆分解释每一部分，最后给出完整的代码段。
在下面的示例中，每个worker的工作都是计算每个数值的位数相加之和。例如给定一个数值234，worker则计算2+3+4=9。这里交给worker的数值是随机生成的[0,999)范围内的数值。
这个示例有几个核心功能需要先解释，也是通过channel实现线程池的一般功能：
首先，创建task和result两个结构，并创建它们的通道：
type task struct {
id int
randnum int
}
type result struct {
task task
result int
}
var tasks = make(chan task, 10)
var results = make(chan result, 10)
这里，每个task都有自己的id，以及该任务将要被worker计算的随机数。每个result都包含了worker的计算结果result以及这个结果对应的task，这样从result中就可以取出任务信息以及计算结果。
另外，两个通道都是buffered channel，容量都是10。每个worker都会监听tasks通道，并取出其中的任务进行计算，然后将计算结果和任务自身放进results通道中。
然后是计算位数之和的函数process()，它将作为worker的工作任务之一。
func process(num int) int {
sum := 0
for num != 0 {
digit := num % 10
sum += digit
num /= 10
}
time.sleep(2 * time.second)
return sum
}
这个计算过程其实很简单，但随后还睡眠了2秒，用来假装执行一个计算任务是需要一点时间的。
然后是worker()，它监听tasks通道并取出任务进行计算，并将结果放进results通道。
func worker(wg *waitgroup){
defer wg.done()
for task := range tasks {
result := result{task, process(task.randnum)}
results <- result
}
}
上面的代码很容易理解，只要tasks channel不关闭，就会一直监听该channel。需要注意的是，该函数使用指针类型的*waitgroup作为参数，不能直接使用值类型的waitgroup作为参数，这样会使得每个worker都有一个自己的waitgroup。
然后是创建工作池的函数createworkerpool()，它有一个数值参数，表示要创建多少个worker。
func createworkerpool(numofworkers int) {
var wg sync.waitgroup
for i := 0; i < numofworkers; i++ {
wg.add(1)
go worker(&wg)
}
wg.wait()
close(results)
}
创建工作池时，首先创建一个waitgroup的值wg，这个wg被工作池中的所有goroutine共享，每创建一个goroutine都wg.add(1)。创建完所有的goroutine后等待所有的groutine都执行完它们的任务，只要有一个任务还没有执行完，这个函数就会被wait()阻塞。当所有任务都执行完成后，关闭results通道，因为没有结果再需要向该通道写了。
当然，这里是否需要关闭results通道，是由稍后的range迭代这个通道决定的，不关闭这个通道会一直阻塞range，最终导致死锁。
工作池部分已经完成了。现在需要使用allocate()函数分配任务：生成一大堆的随机数，然后将task放进tasks通道。该函数有一个代表创建任务数量的数值参数：
func allocate(numoftasks int) {
for i := 0; i < numoftasks; i++ {
randnum := rand.intn(999)
task := task{i, randnum}
tasks <- task
}
close(tasks)
}
注意，最后需要关闭tasks通道，因为所有任务都分配完之后，没有任务再需要分配。当然，这里之所以需要关闭tasks通道，是因为worker()中使用了range迭代tasks通道，如果不关闭这个通道，worker将在取完所有任务后一直阻塞，最终导致死锁。
再接着的是取出results通道中的结果进行输出，函数名为getresult()：
func getresult(done chan bool) {
for result := range results {
fmt.printf("task id %d, randnum %d , sum %d\n", result.task.id, result.task.randnum, result.result)
}
done <- true
}
getresult()中使用了一个done参数，这个参数是一个信号通道，用来表示results中的所有结果都取出来并处理完成了，这个通道不一定要用bool类型，任何类型皆可，它不用来传数据，仅用来返回可读，所以上面直接close(done)的效果也一样。通过下面的main()函数，就能理解done信号通道的作用。
最后还差main()函数：
func main() {
// 记录起始终止时间，用来测试完成所有任务耗费时长
starttime := time.now()
numofworkers := 20
numoftasks := 100
// 创建任务到任务队列中
go allocate(numoftasks)
// 创建工作池
go createworkerpool(numofworkers)
// 取得结果
var done = make(chan bool)
go getresult(done)
// 如果results中还有数据，将阻塞在此
// 直到发送了信号给done通道
<- done
endtime := time.now()
diff := endtime.sub(starttime)
fmt.println("total time taken ", diff.seconds(), "seconds")
}
上面分配了20个worker，这20个worker总共需要处理的任务数量为100。但注意，无论是tasks还是results通道，容量都是10，意味着任务队列最长只能是10个任务。
下面是完整的代码段：
package main
import (
"fmt"
"math/rand"
"sync"
"time"
)
type task struct {
id int
randnum int
}
type result struct {
task task
result int
}
var tasks = make(chan task, 10)
var results = make(chan result, 10)
func process(num int) int {
sum := 0
for num != 0 {
digit := num % 10
sum += digit
num /= 10
}
time.sleep(2 * time.second)
return sum
}
func worker(wg *sync.waitgroup) {
defer wg.done()
for task := range tasks {
result := result{task, process(task.randnum)}
results <- result
}
}
func createworkerpool(numofworkers int) {
var wg sync.waitgroup
for i := 0; i < numofworkers; i++ {
wg.add(1)
go worker(&wg)
}
wg.wait()
close(results)
}
func allocate(numoftasks int) {
for i := 0; i < numoftasks; i++ {
randnum := rand.intn(999)
task := task{i, randnum}
tasks <- task
}
close(tasks)
}
func getresult(done chan bool) {
for result := range results {
fmt.printf("task id %d, randnum %d , sum %d\n", result.task.id, result.task.randnum, result.result)
}
done <- true
}
func main() {
starttime := time.now()
numofworkers := 20
numoftasks := 100
var done = make(chan bool)
go getresult(done)
go allocate(numoftasks)
go createworkerpool(numofworkers)
// 必须在allocate()和getresult()之后创建工作池
<-done
endtime := time.now()
diff := endtime.sub(starttime)
fmt.println("total time taken ", diff.seconds(), "seconds")
}
执行结果：
task id 19, randnum 914 , sum 14 task id 9, randnum 150 , sum 6 task id 15, randnum 215 , sum 8 ............ task id 97, randnum 315 , sum 9 task id 99, randnum 641 , sum 11 total time taken 10.0174705 seconds
总共花费10秒。
可以试着将任务数量、worker数量修改修改，看看它们的性能比例情况。例如，将worker数量设置为99，将需要4秒，将worker数量设置为10，将需要20秒。
如对本文有疑问， 点击进行留言回复！！
网友评论