Golang 学习笔记十三 实例work

摘自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 方法直到所有的工作都做完才会返回。在这个例子里，最多只会等待两个工作的完成。