摘自Go语言实战第7章
work 包的目的是展示如何使用无缓冲的通道来创建一个 goroutine 池，这些 goroutine 执行并控制一组工作，让其并发执行。在这种情况下，使用无缓冲的通道要比随意指定一个缓冲区大小的有缓冲的通道好，因为这个情况下既不需要一个工作队列，也不需要一组 goroutine 配合执行。无缓冲的通道保证两个 goroutine 之间的数据交换。这种使用无缓冲的通道的方法允许使用者知道什么时候 goroutine 池正在执行工作，而且如果池里的所有 goroutine 都忙，无法接受新的工作的时候，也能及时通过通道来通知调用者。使用无缓冲的通道不会有工作在队列里丢失或者卡住，所有工作都会被处理。
//work.go
// Jason Waldrip 协助完成了这个示例
// work 包管理一个 goroutine 池来完成工作
package work
import "sync"
// Worker 必须满足接口类型，
// 才能使用工作池
type Worker interface {
Task()
}
// Pool 提供一个 goroutine 池，这个池可以完成
// 任何已提交的 Worker 任务
type Pool struct {
work chan Worker
wg sync.WaitGroup
}
// New 创建一个新工作池
func New(maxGoroutines int) *Pool {
p := Pool{
work: make(chan Worker),
}
p.wg.Add(maxGoroutines)
for i := 0; i < maxGoroutines; i++ {
go func() {
for w := range p.work {
w.Task()
}
p.wg.Done()
}()
}
return &p
}
// Run 提交工作到工作池
func (p *Pool) Run(w Worker) {
p.work <- w
}
// Shutdown 等待所有 goroutine 停止工作
func (p *Pool) Shutdown() {
close(p.work)
p.wg.Wait()
}
//main.go
// 这个示例程序展示如何使用 work 包
// 创建一个 goroutine 池并完成工作
package main
import (
"log"
"sync"
"time"
"github.com/goinaction/code/chapter7/patterns/work"
)
// names 提供了一组用来显示的名字
var names = []string{
"steve",
"bob",
"mary",
"therese",
"jason",
}
// namePrinter 使用特定方式打印名字
type namePrinter struct {
name string
}
// Task 实现 Worker 接口
func (m *namePrinter) Task() {
log.Println(m.name)
//等待这 1 秒只是为了让测试程序运行的
//速度慢一些，以便看到并发的效果
time.Sleep(time.Second)
}
// main 是所有 Go 程序的入口
func main() {
// 使用两个 goroutine 来创建工作池
p := work.New(2)
var wg sync.WaitGroup
wg.Add(100 * len(names))
for i := 0; i < 100; i++ {
// 迭代 names 切片
for _, name := range names {
// 创建一个 namePrinter 并提供
// 指定的名字
np := namePrinter{
name: name,
}
go func() {
// 将任务提交执行。当 Run 返回时
// 我们就知道任务已经处理完成
p.Run(&np)
wg.Done()
}()
}
}
wg.Wait()
// 让工作池停止工作，等待所有现有的
// 工作完成
p.Shutdown()
}
1.New方法中的for range 循环会一直阻塞，直到从 work 通道收到一个 Worker 接口值。如果收到一个值，就会执行这个值的 Task 方法。一旦 work 通道被关闭，for range循环就会结束，并调用 WaitGroup 的 Done 方法。然后 goroutine 终止。
2.Run 方法可以向池里提交工作。该方法接受一个 Worker类型的接口值作为参数，并将这个值通过 work 通道发送。由于 work 通道是一个无缓冲的通道，调用者必须等待工作池里的某个 goroutine 接收到这个值才会返回。这正是我们想要的，这样可以保证调用的 Run 返回时，提交的工作已经开始执行。
3.Shutdown 方法做了两件事，首先，它关闭了 work 通道，这会导致所有池里的 goroutine 停止工作，并调用 WaitGroup 的 Done 方法；然后，Shutdown 方法调用WaitGroup 的 Wait 方法，这会让 Shutdown 方法等待所有 goroutine 终止。
4.在代码清单 7-36 第 36 行，调用 work 包里的 New 函数创建一个工作池。这个调用传入的参数是 2，表示这个工作池只会包含两个执行任务的 goroutine。在第 38 行和第 39 行，声明了一个
WaitGroup，并初始化为要执行任务的 goroutine 数。在这个例子里，names 切片里的每个名字都会创建 100 个 goroutine 来提交任务。这样就会有一堆 goroutine 互相竞争，将任务提交到池里。
在第 41 行到第 43 行，内部和外部的 for 循环用来声明并创建所有的 goroutine。每次内部循环都会创建一个 namePrinter 类型的值，并提供一个用来打印的名字。之后，在第 50 行，
声明了一个匿名函数，并创建一个 goroutine 执行这个函数。这个 goroutine 会调用工作池的 Run方法，将 namePrinter 的值提交到池里。一旦工作池里的 goroutine 接收到这个值，Run 方法
就会返回。这也会导致 goroutine 将 WaitGroup 的计数递减，并终止 goroutine。一旦所有的 goroutine 都创建完成，main 函数就会调用 WaitGroup 的 Wait 方法。这个调用会等待所有创建的 goroutine 提交它们的工作。一旦 Wait 返回，就会调用工作池的 Shutdown方法来关闭工作池。Shutdown 方法直到所有的工作都做完才会返回。在这个例子里，最多只会等待两个工作的完成。
网友评论