第22课：Channel 模式与实践

学习目标

理解并掌握使用 Channel 实现 Pipeline（管道） 模式，构建可组合、可扩展的数据处理流水线。
学会使用 Fan-out（扇出） 模式将任务分发给多个 goroutine 以并行处理，提高吞吐量。
学会使用 Fan-in（扇入） 模式将多个 channel 的结果汇聚到一个 channel 中，简化结果收集。
了解带缓冲的 Channel (Buffered Channel) 在特定模式下的应用与优势。
能够综合运用上述模式，设计出高效的并发数据处理方案。

核心概念

在上一课中，我们学习了 Channel 的基础用法，它用于在 goroutine 之间安全地传递数据。今天，我们将探索几种由 Channel 组合而成的经典并发设计模式，它们是构建强大并发程序的基石。

1. Pipeline（管道）

Pipeline 模式就像一条工厂的生产线。原材料（数据）通过一系列 阶段 (stage) goroutine 进行处理，每个阶段执行特定的变换操作，最后输出成品。

阶段 (Stage): 一个 goroutine，它从一个输入 channel 接收数据，处理数据，然后将结果发送到一个输出 channel。
连接: 前一个阶段的输出 channel 成为后一个阶段的输入 channel。

比喻: 厨师做菜。生成食材 goroutine -> 切菜 goroutine -> 炒菜 goroutine -> 装盘 goroutine。每个环节只做一件事，通过“传递窗口”（channel）连接起来。

2. Fan-out / Fan-in（扇出/扇入）

这两个模式经常配合使用，用于处理 pipeline 中的“瓶颈”阶段。

Fan-out (扇出): 多个 goroutine 从同一个 channel 读取数据，实现对该阶段的并行处理。适用于某个阶段的处理速度是瓶颈的情况。
Fan-in (扇入): 多个 goroutine 向同一个 channel 写入数据，或者一个 goroutine 从多个 channel 中读取数据并将它们合并到一个输出 channel。常用于汇总多个并行任务的结果。

比喻:

扇出: 传菜窗口只有一个，但有多个厨师（goroutine）同时从这里取菜去烹饪。
扇入: 多个厨师做完菜后，都把成品放到同一个传菜窗口，由一个服务员统一送给顾客。

3. Buffered Channel（带缓冲的 Channel）

在上述模式中，如果使用无缓冲 channel，发送方和接收方必须同时准备好才能进行通信，这可能导致不必要的等待。带缓冲的 channel 允许发送方在接收方准备好之前发送一定数量的数据。

作用：减少 goroutine 之间的同步阻塞，提高某些场景下的吞吐量。
注意：缓冲区满时发送会阻塞，缓冲区空时接收会阻塞。它不能用于保证数据的有序性，只是一种优化手段。

代码示例

下面是一个综合示例，演示了 Pipeline 和 Fan-out/Fan-in 模式。我们将构建一个文本处理流水线，它接收一个句子列表，转换大小写，并添加前缀。

package main

import (
	"fmt"
	"strings"
	"sync"
)

// 生成器：将数据发送到 channel
func generator(input ...string) <-chan string {
	out := make(chan string)
	go func() {
		for _, s := range input {
			out <- s
		}
		close(out)
	}()
 return out
}

// 阶段1：转换为大写 (Pipeline Stage)
func toUpper(in <-chan string) <-chan string {
	out := make(chan string)
	go func() {
		for s := range in {
			out <- strings.ToUpper(s)
		}
		close(out)
	}()
	return out
}

// 阶段2：添加前缀 (Pipeline Stage)
func addPrefix(prefix string, in <-chan string) <-chan string {
	out := make(chan string)
	go func() {
		for s := range in {
			out <- prefix + s
		}
		close(out)
	}()
	return out
}

// Fan-out Worker：模拟一个需要并行处理的耗时操作（例如：格式化）
func formatWorker(id int, in <-chan string, out chan<- string, wg *sync.WaitGroup) {
	defer wg.Done()
	for s := range in {
		// 模拟耗时操作
		fmt.Printf("Worker %d processing: %s\n", id, s)
		out <- fmt.Sprintf("[%d] %s", id, s)
	}
}

// Fan-in：合并多个 channel 的结果
func fanIn(inputChannels ...<-chan string) <-chan string {
	var wg sync.WaitGroup
	merged := make(chan string)
	wg.Add(len(inputChannels))

	// 为每个输入 channel 启动一个转发 goroutine
	for _, ch := range inputChannels {
		go func(c <-chan string) {
			defer wg.Done()
			for val := range c {
				merged <- val
			}
		}(ch)
	}

	// 等待所有转发 goroutine 结束后关闭 merged channel
	go func() {
		wg.Wait()
		close(merged)
	}()
	return merged
}

func main() {
	// 1. 生成原始数据
	sentences := []string{
		"hello world",
		"Go concurrency is powerful",
		"learn patterns",
		"practice makes perfect",
	}
	source := generator(sentences...)

	// 2. 构建 Pipeline：生成器 -> 转大写 -> 添加前缀
	upperCase := toUpper(source)
	prefixed := addPrefix("PROCESSED: ", upperCase)

	// 3. Fan-out: 将 `prefixed` channel 的任务分发给3个 worker 并行处理
	workerCount := 3
	workerChannels := make([]<-chan string, workerCount)
	var wg sync.WaitGroup
	wg.Add(workerCount)

	for i := 0; i < workerCount; i++ {
		// 为每个 worker 创建一个输出 channel
		out := make(chan string)
		workerChannels[i] = out
		go formatWorker(i+1, prefixed, out, &wg)
	}

	// 启动一个 goroutine 来等待所有 worker 完成，然后关闭它们的输出 channel
	go func() {
		wg.Wait()
		for _, ch := range workerChannels {
			// 注意：这里的 ch 是只读 channel (interface)，我们无法关闭它。
			// 我们已经通过传递一个有明确类型的输出 channel (`out chan<- string`) 给 worker，
			// worker 的 defer wg.Done() 之后，该 channel 就不再被写入了。
			// 我们需要一种方式来关闭它，但这里为了示例简洁，我们依赖 `fanIn` 函数内部的转发 goroutine 最终会因输入 channel 关闭而结束。
		}
	}()

	// 4. Fan-in: 将所有 worker 的结果合并到一个 channel
	merged := fanIn(workerChannels...)

	// 5. 从最终的 merged channel 中读取并打印结果
	fmt.Println("--- Final Results ---")
	for result := range merged {
		fmt.Println(result)
	}
	fmt.Println("--- All Done ---")
}

实践练习

练习1：基础 Pipeline

创建一个简单的 pipeline，包含两个阶段：

生成 1 到 10 的数字。
计算每个数字的平方。
主函数从最终 channel 中打印所有结果。要求：使用 Pipeline 模式，每个阶段是独立的函数。

练习2：带 Fan-out 的 Pipeline

在练习1的基础上，增加一个“打印”阶段。但是，“打印”阶段模拟为耗时操作（例如，time.Sleep(100*time.Millisecond)）。使用 Fan-out 模式，启动 3 个 goroutine 并行执行“打印”任务，以加快整体速度。

练习3：综合应用 - 文档单词统计

设计一个并发程序来统计多个文档中单词的出现频率。

输入：一个字符串切片，每个字符串是一段文档内容。
要求：
1. 使用一个 goroutine 作为生成器，将文档内容逐个发送到 channel。
2. 使用 Fan-out 启动多个 worker goroutine，每个 worker 接收一个文档内容，统计其中每个单词的出现次数，并将结果（map[string]int）发送到自己的输出 channel。
3. 使用 Fan-in 将所有 worker 的输出合并。
4. 主 goroutine 将所有的局部 map 合并成一个全局的单词频率 map。
5. 打印出最终的单词频率 map。

常见错误

忘记关闭 Channel：在 Pipeline 和 Fan-in 模式中，如果生产者 goroutine 完成后不关闭 channel，消费者 goroutine 的 for range 循环将永远阻塞，导致 goroutine 泄漏和死锁。
- 解决：确保每个负责“结束”数据流的 goroutine 在其任务完成后调用 close(channel)。
向已关闭的 Channel 发送数据：这会导致 panic。务必确保数据发送完成后才关闭 channel，并且只有一个发送方负责关闭。
- 解决：设计清晰的 channel 所有权和生命周期。通常由创建 channel 的 goroutine 或负责最后一份数据的发送方来关闭。
在 Fan-out 中错误地共享 channel：如果多个 goroutine 向同一个无缓冲的 channel 写入数据，没有同步控制，虽然操作是安全的，但可能无法控制任务的均匀分发。
- 解决：让每个 worker 从同一个输入 channel 读取，由 Go runtime 自动进行负载均衡。这是标准的 Fan-out 做法。
误用 Buffered Channel 作为同步原语：Buffered channel 可以减少阻塞，但不能替代 WaitGroup 或 Mutex 来保证特定的执行顺序或保护共享内存。
- 解决：Buffered channel 主要用于解耦生产者和消费者的速率，而不是用于精确的并发控制。

小结

Pipeline 通过将任务分解为一系列阶段，并用 channel 连接，使代码更清晰、可复用。
Fan-out 利用多个 goroutine 从同一 channel 读取，实现并行处理，提升 CPU 密集型或 I/O 密集型任务的吞吐量。
Fan-in 汇聚多个 channel 的结果，简化了并行任务结果的收集过程。
Buffered Channel 在生产/消费速度不匹配时能提供缓冲，减少阻塞，是性能优化的工具。
设计这些模式时，Channel 的所有权和生命周期管理（特别是关闭时机） 至关重要，需要仔细设计以避免死锁和 panic。

掌握这些模式，你就能像搭积木一样，用 goroutine 和 channel 构建出复杂而高效的并发系统。下一课，我们将学习 select 语句，它能让你同时监听多个 channel，是实现更复杂并发逻辑的关键。

Channel 模式与实践