小项目：流水线和工人

第 24 课，共 24 课 · 第 6 部分 · 小项目 · 约 20 分钟

这一课把并发的几课合起来，一步步搭两个常见的程序结构：

流水线：数据一站一站往下走。每一站是一个 goroutine，站和站之间用 channel 连起来。
一群工人（worker pool）：很多活放进一个 channel，几个工人 goroutine 谁有空谁去拿，做完把结果交回来。

我们会特别小心一件事：不管 goroutine 谁先谁后，显示的结果都要一样。

第一站：生成

流水线的第一站把一串价格放进 channel。它是一个函数：开一个 goroutine 去发，自己马上把 channel 返回：

prog.go

1package main2 3import "fmt"4 5func generate(prices ...int) <-chan int {6	out := make(chan int)7	go func() {8		defer close(out)9		for _, p := range prices {10			out <- p11		}12	}()13	return out14}15 16func main() {17	for p := range generate(120, 45, 80) {18		fmt.Println(p)19	}20}

输出

120
45
80

返回类型 <-chan int 是「只能收的 channel」（第 19 课）：拿到它的人只能收，不能发，也不能关。发的是 generate 里的 goroutine，所以也由它 close：谁发，谁关。

第二站：打折

第二站从上一站收价格，打九折，再发给下一站。它的参数是一个只能收的 channel，返回一个新的：

prog.go

1package main2 3import "fmt"4 5func generate(prices ...int) <-chan int {6	out := make(chan int)7	go func() {8		defer close(out)9		for _, p := range prices {10			out <- p11		}12	}()13	return out14}15 16func discount(in <-chan int) <-chan int {17	out := make(chan int)18	go func() {19		defer close(out)20		for p := range in {21			out <- p * 9 / 1022		}23	}()24	return out25}26 27func main() {28	for p := range discount(generate(120, 45, 80)) {29		fmt.Println(p)30	}31}

输出

108
40
72

点单步：三条泳道，generate、discount 和 main。一个价格从第一条泳道发出，discount 收到、算好、再发出，main 收到后显示。上一站 close 了，discount 的 range 就结束，它也 close 自己的 channel，一站一站往下传。

每一站只有一个 goroutine，channel 先发的先到，所以价格的顺序一路都不变。

第三站：汇总

最后一站把所有价格加起来。它只发一个值：合计。

prog.go

1package main2 3import "fmt"4 5func generate(prices ...int) <-chan int {6	out := make(chan int)7	go func() {8		defer close(out)9		for _, p := range prices {10			out <- p11		}12	}()13	return out14}15 16func discount(in <-chan int) <-chan int {17	out := make(chan int)18	go func() {19		defer close(out)20		for p := range in {21			out <- p * 9 / 1022		}23	}()24	return out25}26 27func sum(in <-chan int) <-chan int {28	out := make(chan int)29	go func() {30		total := 031		for p := range in {32			total += p33		}34		out <- total35	}()36	return out37}38 39func main() {40	fmt.Println("合计", <-sum(discount(generate(120, 45, 80))))41}

输出

合计 220

sum(discount(generate(…))) 从里往外读：生成，打折，汇总。main 只做一件事：从最后一站收合计。

再加一站

流水线的好处是好加站。比如只给 50 元以上的商品打折，便宜的不算进去：加一个筛选站 atLeast，接在 generate 和 discount 中间就行，别的站一行都不用改。

prog.go

1package main2 3import "fmt"4 5func generate(prices ...int) <-chan int {6	out := make(chan int)7	go func() {8		defer close(out)9		for _, p := range prices {10			out <- p11		}12	}()13	return out14}15 16func atLeast(in <-chan int, limit int) <-chan int {17	out := make(chan int)18	go func() {19		defer close(out)20		for p := range in {21			if p >= limit {22				out <- p23			}24		}25	}()26	return out27}28 29func discount(in <-chan int) <-chan int {30	out := make(chan int)31	go func() {32		defer close(out)33		for p := range in {34			out <- p * 9 / 1035		}36	}()37	return out38}39 40func main() {41	for p := range discount(atLeast(generate(120, 45, 80, 30), 50)) {42		fmt.Println(p)43	}44}

输出

108
72

先猜一猜

在真机上，这段程序显示什么？

1package main2 3import "fmt"4 5func generate(nums ...int) <-chan int {6	out := make(chan int)7	go func() {8		defer close(out)9		for _, n := range nums {10			out <- n11		}12	}()13	return out14}15 16func double(in <-chan int) <-chan int {17	out := make(chan int)18	go func() {19		defer close(out)20		for n := range in {21			out <- n * 222		}23	}()24	return out25}26 27func main() {28	for n := range double(double(generate(1, 2, 3))) {29		fmt.Print(n, " ")30	}31	fmt.Println()32}

double 接了两次，每个数乘了 4。每一站只有一个 goroutine，channel 又是先发先到，所以顺序一定不变。

输出

一群工人

流水线的每一站只有一个 goroutine。如果某一站很慢（比如做缩略图），可以让几个工人一起做。下面 3 个工人从 jobs 里拿图片，做好的缩略图放进 results：

prog.go

1package main2 3import (4	"fmt"5	"sort"6	"sync"7)8 9type Image struct {10	ID   int11	Size int12}13 14func worker(jobs <-chan Image, results chan<- Image, wg *sync.WaitGroup) {15	defer wg.Done()16	for img := range jobs {17		results <- Image{ID: img.ID, Size: img.Size / 10}18	}19}20 21func main() {22	sizes := []int{1200, 800, 2000, 640, 1500}23	jobs := make(chan Image)24	results := make(chan Image)25	var wg sync.WaitGroup26	for range 3 {27		wg.Add(1)28		go worker(jobs, results, &wg)29	}30	go func() {31		defer close(jobs)32		for i, s := range sizes {33			jobs <- Image{ID: i + 1, Size: s}34		}35	}()36	go func() {37		wg.Wait()38		close(results)39	}()40	var thumbs []Image41	for t := range results {42		thumbs = append(thumbs, t)43	}44	sort.Slice(thumbs, func(a, b int) bool { return thumbs[a].ID < thumbs[b].ID })45	for _, t := range thumbs {46		fmt.Printf("图片 %d：%d KB\n", t.ID, t.Size)47	}48}

输出

图片 1：120 KB
图片 2：80 KB
图片 3：200 KB
图片 4：64 KB
图片 5：150 KB

部分	做什么
`jobs`	要做的活。一个 goroutine 把图片都放进去，放完 `close`。
`worker`	每个工人 `range jobs`，拿一张做一张，`jobs` 关了、拿完了就结束。
`results`	做好的缩略图。
`wg.Wait()` 再 `close(results)`	所有工人都结束了，才关 `results`。这件事放在单独的 goroutine 里，`main` 好一边去收结果。
`sort.Slice`	收齐以后按图片编号排好，再显示。

点单步：看三条工人泳道怎样从 jobs 拿活，main 又在等哪个 channel。

为什么要排序？哪个工人先做完是说不准的。如果收到一个就显示一个：

1for t := range results {2	fmt.Printf("图片 %d：%d KB\n", t.ID, t.Size)3}

在真机上，这次可能是 1、3、2、4、5，下次是 2、1、3、5、4。我们的网页一次只跑一个 goroutine，顺序每次都一样，但别依赖它。先全部收齐，再排序，结果就和谁先谁后没关系了。同样，也别显示「哪个工人做了哪张图」：那每次都可能不一样。

工人多少，结果一样

收集结果还有一种办法：不用 results channel，每个工人把结果写进自己那张图片的格子 thumbs[i]（第 22 课）。下面把整个工人池写成一个函数，分别用 1、2、5 个工人运行：

prog.go

1package main2 3import (4	"fmt"5	"sync"6)7 8func makeThumbs(sizes []int, workers int) []int {9	thumbs := make([]int, len(sizes))10	jobs := make(chan int)11	var wg sync.WaitGroup12	for range workers {13		wg.Add(1)14		go func() {15			defer wg.Done()16			for i := range jobs {17				thumbs[i] = sizes[i] / 1018			}19		}()20	}21	for i := range sizes {22		jobs <- i23	}24	close(jobs)25	wg.Wait()26	return thumbs27}28 29func main() {30	sizes := []int{1200, 800, 2000, 640, 1500}31	for _, n := range []int{1, 2, 5} {32		fmt.Printf("%d 个工人：%v\n", n, makeThumbs(sizes, n))33	}34}

输出

1 个工人：[120 80 200 64 150]
2 个工人：[120 80 200 64 150]
5 个工人：[120 80 200 64 150]

jobs 里放的是下标。每个下标只会被一个工人拿到，所以每一格只有一个工人写，不用加锁。工人数只影响谁做哪张，不影响结果。

合在一起

最后把两种结构接起来：生成 → 3 个工人一起打折 → 汇总。打折这一站变成几个工人，出来的顺序就说不准了；可是汇总只是相加，顺序不影响合计。

prog.go

1package main2 3import (4	"fmt"5	"sync"6)7 8func generate(prices ...int) <-chan int {9	out := make(chan int)10	go func() {11		defer close(out)12		for _, p := range prices {13			out <- p14		}15	}()16	return out17}18 19func discountWorkers(in <-chan int, n int) <-chan int {20	out := make(chan int)21	var wg sync.WaitGroup22	for range n {23		wg.Add(1)24		go func() {25			defer wg.Done()26			for p := range in {27				out <- p * 9 / 1028			}29		}()30	}31	go func() {32		wg.Wait()33		close(out)34	}()35	return out36}37 38func sum(in <-chan int) <-chan int {39	out := make(chan int)40	go func() {41		total := 042		for p := range in {43			total += p44		}45		out <- total46	}()47	return out48}49 50func main() {51	prices := discountWorkers(generate(120, 45, 80, 300, 15), 3)52	fmt.Println("合计", <-sum(prices))53}

输出

合计 503

几个工人往同一个 out 发，谁也不能先关它：等 wg.Wait() 确认都结束了，再由一个 goroutine 关。这和上面的 results 是同一个办法。

练一练

写好代码，点检查。提示随便看，实在想不出来再看答案。

练习 1

流水线加一站：运费

每件商品打折以后再加 5 元运费。写一个新站 shipping(in <-chan int, fee int) <-chan int，把它接在 discount 和 sum 中间。程序算了两个订单。

要显示

195
217

exercise1.go

1package main2 3import "fmt"4 5func generate(prices ...int) <-chan int {6	out := make(chan int)7	go func() {8		defer close(out)9		for _, p := range prices {10			out <- p11		}12	}()13	return out14}15 16func discount(in <-chan int) <-chan int {17	out := make(chan int)18	go func() {19		defer close(out)20		for p := range in {21			out <- p * 9 / 1022		}23	}()24	return out25}26 27// 在这里写 shipping28 29func sum(in <-chan int) <-chan int {30	out := make(chan int)31	go func() {32		total := 033		for p := range in {34			total += p35		}36		out <- total37	}()38	return out39}40 41func main() {42	fmt.Println(<-sum(discount(generate(100, 40, 60))))43	fmt.Println(<-sum(discount(generate(200, 30))))44}

输出

点「运行」或按 Ctrl+Enter

练习 2

几个工人一起算含税价

withTax 把每个价格加上 10% 的税。活已经往 jobs 里放了，可是还没有工人。开 workers 个工人 goroutine，每个从 jobs 拿下标，把含税价写进 out 的那一格。2 个工人和 4 个工人算出来的要一样。

要显示

[110 275 44 77]
[110 275 44 77]

exercise2.go

1package main2 3import (4	"fmt"5	"sync"6)7 8func withTax(prices []int, workers int) []int {9	out := make([]int, len(prices))10	jobs := make(chan int)11	var wg sync.WaitGroup12	// 在这里开工人13 14	for i := range prices {15		jobs <- i16	}17	close(jobs)18	wg.Wait()19	return out20}21 22func main() {23	prices := []int{100, 250, 40, 70}24	fmt.Println(withTax(prices, 2))25	fmt.Println(withTax(prices, 4))26}

输出

点「运行」或按 Ctrl+Enter

练习 3

修好它：结果没人收的打包工人

3 个工人给订单算要几个箱子（一个箱子装 10 件），可是程序死锁了：pack 先 wg.Wait() 等工人结束，工人却卡在往 results 发结果上。改一改，让 pack 一边收结果，另一个 goroutine 等工人结束再关 results。程序打包了两批订单。

要显示

订单 1：3 箱
订单 2：1 箱
订单 3：1 箱
订单 4：5 箱
订单 5：6 箱
订单 6：1 箱
订单 7：2 箱

exercise3.go

1package main2 3import (4	"fmt"5	"sort"6	"sync"7)8 9type Order struct {10	ID    int11	Items int12}13 14func pack(orders []Order) {15	jobs := make(chan Order)16	results := make(chan Order)17	var wg sync.WaitGroup18	for range 3 {19		wg.Add(1)20		go func() {21			defer wg.Done()22			for o := range jobs {23				results <- Order{ID: o.ID, Items: (o.Items + 9) / 10}24			}25		}()26	}27	go func() {28		defer close(jobs)29		for _, o := range orders {30			jobs <- o31		}32	}()33	wg.Wait()34	close(results)35	var boxes []Order36	for r := range results {37		boxes = append(boxes, r)38	}39	sort.Slice(boxes, func(a, b int) bool { return boxes[a].ID < boxes[b].ID })40	for _, b := range boxes {41		fmt.Printf("订单 %d：%d 箱\n", b.ID, b.Items)42	}43}44 45func main() {46	pack([]Order{47		{ID: 1, Items: 25},48		{ID: 2, Items: 10},49		{ID: 3, Items: 3},50		{ID: 4, Items: 41},51	})52	pack([]Order{53		{ID: 5, Items: 60},54		{ID: 6, Items: 9},55		{ID: 7, Items: 11},56	})57}

输出

点「运行」或按 Ctrl+Enter

这一课学会了

流水线：每一站是一个函数，收一个 <-chan，开一个 goroutine 处理，返回一个新的 <-chan；站和站用函数调用接起来。
每一站发完就 defer close(out)，下一站的 range 才会结束。
每站一个 goroutine 时，channel 先发先到，顺序不变；加站不用改别的站。
工人池：活放进 jobs，几个工人 range jobs；所有工人结束（wg.Wait()）以后，由一个单独的 goroutine 关 results。
工人做完的先后说不准：先收齐再排序，或者每个工人写自己的格子，结果就和调度无关。
汇总（相加、计数）不在乎顺序，可以直接接在工人后面。