Go 并发模式：Pipeline 与取消

大家好，我是江湖十年。

Go 官方博客在 2014 年发表过一篇介绍并发流水线的文章《Go Concurrency Patterns: Pipelines and cancellation》。文中通过逐步演进的示例，展示了如何用 channel 构建流式数据处理流水线，以及在操作失败、提前退出等场景下如何优雅地取消、避免 goroutine 泄漏。本文是对这篇文章的中文翻译。

如下为译文：

原文地址：https://go.dev/blog/pipelines

Sameer Ajmani

2014 年 3 月 13 日

引言

Go 的并发原语让构建流式数据处理流水线变得轻而易举，这类流水线能够高效利用 I/O 和多个 CPU。本文给出了此类流水线的示例，重点剖析了操作失败时出现的微妙问题，并介绍了干净利落地处理失败的技术。

什么是流水线？

Go 中并没有关于流水线的正式定义；它只是众多并发程序中的一种。简单来说，流水线是由 channel 连接起来的一系列阶段（stage），每个阶段都是一组运行同一函数的 goroutines。在每个阶段中，goroutines 会：

• 通过入站（inbound）channel 从上游（upstream）接收值
• 对这些数据执行某种函数操作，通常会生成新的值
• 通过出站（outbound）channel 将值发送到下游（downstream）

每个阶段可以有任意数量的入站和出站 channel，但第一个阶段和最后一个阶段分别只有出站和入站 channel。第一个阶段有时被称为源（source）或生产者（producer）；最后一个阶段则被称为接收端（sink）或消费者（consumer）。

我们先从一个简单的示例流水线开始，讲解相关思想和技术。稍后，再给出一个更贴近实际的例子。

对数字求平方

考虑一个包含三个阶段的流水线。

第一个阶段 gen 是一个函数，它把一组整数转换成一个 channel，该 channel 会逐个发出这组整数。gen 函数会启动一个 goroutine，在 channel 上发送这些整数，并在所有值发送完毕后关闭 channel：

func gen(nums ...int) <-chan int {
	out := make(chan int)
	go func() {
		for _, n := range nums {
			out <- n
		}
		close(out)
	}()
	return out
}

第二个阶段 sq 从 channel 接收整数，并返回一个 channel，该 channel 会发出每个收到整数的平方。当入站 channel 关闭、且该阶段已将所有值发送到下游后，它会关闭出站 channel：

func sq(in <-chan int) <-chan int {
	out := make(chan int)
	go func() {
		for n := range in {
			out <- n * n
		}
		close(out)
	}()
	return out
}

main 函数搭建好流水线并运行最后一个阶段：它从第二个阶段接收值并逐个打印，直到 channel 关闭：

func main() {
	// Set up the pipeline.
	c := gen(2, 3)
	out := sq(c)

	// Consume the output.
	fmt.Println(<-out) // 4
	fmt.Println(<-out) // 9
}

由于 sq 的入站 channel 和出站 channel 类型相同，我们可以将它任意组合多次。我们也可以像其他阶段一样，把 main 改写成 range 循环：

func main() {
	// Set up the pipeline and consume the output.
	for n := range sq(sq(gen(2, 3))) {
		fmt.Println(n) // 16 then 81
	}
}

扇出、扇入

多个函数可以从同一个 channel 读取数据，直到该 channel 关闭；这被称为扇出（fan-out）。它提供了一种在一组 worker 之间分配工作的方法，从而并行利用 CPU 和 I/O。

一个函数可以从多个输入读取数据，并通过将输入 channel 多路复用到单个 channel 上来持续处理，直到所有输入都关闭，而这个输出 channel 会在所有输入关闭后关闭。这被称为扇入（fan-in）。

我们可以修改流水线，运行两个 sq 实例，它们都从同一个输入 channel 读取数据。为此引入一个新函数 merge，用来将结果扇入：

func main() {
	in := gen(2, 3)

	// Distribute the sq work across two goroutines that both read from in.
	c1 := sq(in)
	c2 := sq(in)

	// Consume the merged output from c1 and c2.
	for n := range merge(c1, c2) {
		fmt.Println(n) // 4 then 9, or 9 then 4
	}
}

merge 函数会为每个入站 channel 启动一个 goroutine，由它把值复制到唯一的出站 channel，从而将一组 channel 转换为单个 channel。所有 output goroutines 启动后，merge 再启动一个 goroutine，在该 channel 上的所有发送完成后关闭出站 channel。

向已关闭的 channel 发送数据会 panic，因此必须确保在调用 close 之前所有发送都已完成。sync.WaitGroup 类型提供了一种简单的方式来安排这种同步：

func merge(cs ...<-chan int) <-chan int {
	var wg sync.WaitGroup
	out := make(chan int)

	// Start an output goroutine for each input channel in cs.  output
	// copies values from c to out until c is closed, then calls wg.Done.
	output := func(c <-chan int) {
		for n := range c {
			out <- n
		}
		wg.Done()
	}
	wg.Add(len(cs))
	for _, c := range cs {
		go output(c)
	}

	// Start a goroutine to close out once all the output goroutines are
	// done.  This must start after the wg.Add call.
	go func() {
		wg.Wait()
		close(out)
	}()
	return out
}

提前停止

我们的流水线函数遵循一个模式：

• 各阶段在所有发送操作完成后关闭自己的出站 channel。
• 各阶段持续从入站 channel 接收值，直到这些 channel 关闭。

这个模式让每个接收阶段都能写成 range 循环，并确保一旦所有值成功发送到下游，所有 goroutines 都会退出。

但在真实的流水线中，阶段并不总能接收全部入站值。有时这是有意为之：接收方可能只需要一部分值就能继续推进。更常见的情况是，某个阶段因为某个入站值代表上游阶段的错误而提前退出。无论哪种情况，接收方都不应被迫等待剩余的值到达，而且我们希望上游阶段停止生产下游不再需要的值。

在我们的示例流水线中，如果某个阶段未能消费全部入站值，那么尝试发送这些值的 goroutines 将永久阻塞：

	// Consume the first value from the output.
	out := merge(c1, c2)
	fmt.Println(<-out) // 4 or 9
	return
	// Since we didn't receive the second value from out,
	// one of the output goroutines is hung attempting to send it.
}

这是一种资源泄漏：goroutines 会占用内存和运行时资源，goroutine 栈中的堆引用还会阻止数据被垃圾回收。Goroutines 不会被垃圾回收；它们必须自行退出。

我们需要让流水线的上游阶段即使在下游阶段未能接收全部入站值时也能退出。一种方法是给出站 channel 增加缓冲区。缓冲区可以容纳固定数量的值；只要缓冲区还有空间，发送操作就会立即完成：

c := make(chan int, 2) // buffer size 2
c <- 1  // succeeds immediately
c <- 2  // succeeds immediately
c <- 3  // blocks until another goroutine does <-c and receives 1

当发送的值数量在创建 channel 时就已知时，缓冲区可以简化代码。例如，我们可以重写 gen，将整数列表复制到带缓冲的 channel 中，从而避免创建新的 goroutine：

func gen(nums ...int) <-chan int {
	out := make(chan int, len(nums))
	for _, n := range nums {
		out <- n
	}
	close(out)
	return out
}

回到流水线中被阻塞的 goroutines，我们或许可以考虑给 merge 返回的出站 channel 添加一个缓冲区：

func merge(cs ...<-chan int) <-chan int {
	var wg sync.WaitGroup
	out := make(chan int, 1) // enough space for the unread inputs
	// ... the rest is unchanged ...

虽然这修复了本程序中被阻塞的 goroutine，但这是糟糕的代码。这里选择缓冲区大小为 1，取决于预先知道 merge 会接收多少个值，以及下游阶段会消费多少个值。这很脆弱：如果我们向 gen 多传一个值，或者下游阶段少读一个值，我们又会遇到被阻塞的 goroutines。

因此，我们需要提供一种方式，让下游阶段告诉发送方它们将停止接收输入。

显式取消

当 main 决定不接收 out 中的全部值就退出时，它必须告诉上游阶段的 goroutines 放弃它们正在尝试发送的值。为此，它在一个名为 done 的 channel 上发送值。由于可能存在两个被阻塞的发送方，它发送两个值：

func main() {
	in := gen(2, 3, 5)

	// Distribute the sq work across two goroutines that both read from in.
	c1 := sq(in)
	c2 := sq(in)

	// Consume the first value from output.
	done := make(chan struct{}, 2)
	out := merge(done, c1, c2)
	fmt.Println(<-out) // 4, 9, or 25

	// Tell the remaining senders we're leaving.
	done <- struct{}{}
	done <- struct{}{}
}

发送方 goroutines 会把发送操作替换为 select 语句，该语句要么在 out 上的发送成功时继续，要么在从 done 接收到值时继续。done 的值类型是空结构体，因为这个值本身并不重要：真正指示应放弃 out 上发送操作的是接收事件。output goroutines 会继续在入站 channel c 上循环，因此上游阶段不会被阻塞。（稍后我们会讨论如何让这个循环提前返回。）

func merge(done <-chan struct{}, cs ...<-chan int) <-chan int {
	var wg sync.WaitGroup
	out := make(chan int)

	// Start an output goroutine for each input channel in cs.  output
	// copies values from c to out until c is closed or it receives a value
	// from done, then output calls wg.Done.
	output := func(c <-chan int) {
	loop:
		for n := range c {
			select {
			case out <- n:
			case <-done:
				break loop
			}
		}
		wg.Done()
	}
	// ... the rest is unchanged ...

这种方法有个问题：每个下游接收方都需要知道可能被阻塞的上游发送方的数量，并在提前返回时设法通知这些发送方。跟踪这些计数既繁琐又容易出错。

我们需要一种方式，告诉数量未知且无上限的 goroutines 停止向下游发送值。在 Go 中，我们可以通过关闭 channel 来做到这一点，因为对已关闭 channel 的接收操作总是可以立即进行，并返回元素类型的零值。

这意味着 main 只需关闭 done channel 就能解除所有发送方的阻塞。这次关闭实际上就是向所有发送方广播信号。我们扩展每个流水线函数，使其接受 done 作为参数，并通过 defer 语句安排关闭操作，这样 main 的所有返回路径都会通知流水线各阶段退出。

func main() {
	// Set up a done channel that's shared by the whole pipeline,
	// and close that channel when this pipeline exits, as a signal
	// for all the goroutines we started to exit.
	done := make(chan struct{})
	defer close(done)

	in := gen(done, 2, 3)

	// Distribute the sq work across two goroutines that both read from in.
	c1 := sq(done, in)
	c2 := sq(done, in)

	// Consume the first value from output.
	out := merge(done, c1, c2)
	fmt.Println(<-out) // 4 or 9

	// done will be closed by the deferred call.
}

现在，流水线的每个阶段都可以在 done 关闭后立即返回。merge 中的 output 例程无需排空入站 channel 就可以返回，因为它知道上游发送方 sq 在 done 关闭时会停止尝试发送。output 通过 defer 语句确保在所有返回路径上都会调用 wg.Done：

func merge(done <-chan struct{}, cs ...<-chan int) <-chan int {
	var wg sync.WaitGroup
	out := make(chan int)

	// Start an output goroutine for each input channel in cs.  output
	// copies values from c to out until c or done is closed, then calls
	// wg.Done.
	output := func(c <-chan int) {
		defer wg.Done()
		for n := range c {
			select {
			case out <- n:
			case <-done:
				return
			}
		}
	}
	// ... the rest is unchanged ...

同样，sq 也可以在 done 关闭后立即返回。sq 通过 defer 语句确保在所有返回路径上关闭其 out channel：

func sq(done <-chan struct{}, in <-chan int) <-chan int {
	out := make(chan int)
	go func() {
		defer close(out)
		for n := range in {
			select {
			case out <- n * n:
			case <-done:
				return
			}
		}
	}()
	return out
}

以下是构建流水线的指导原则：

• 各阶段在所有发送操作完成后关闭自己的出站 channel。
• 各阶段持续从入站 channel 接收值，直到这些 channel 关闭或发送方被解除阻塞。

流水线要么通过确保缓冲区足以容纳所有发送的值来解除发送方的阻塞，要么在接收方可能放弃 channel 时显式通知发送方。

对目录树做摘要

我们来看一个更贴近实际的流水线。

MD5 是一种消息摘要算法，可用作文件校验和。命令行工具 md5sum 会为一组文件打印摘要值。

% md5sum *.go
d47c2bbc28298ca9befdfbc5d3aa4e65  bounded.go
ee869afd31f83cbb2d10ee81b2b831dc  parallel.go
b88175e65fdcbc01ac08aaf1fd9b5e96  serial.go

我们的示例程序类似 md5sum，但它只接收一个目录作为参数，并打印该目录下每个常规文件的摘要值，结果按路径名排序。

% go run serial.go .
d47c2bbc28298ca9befdfbc5d3aa4e65  bounded.go
ee869afd31f83cbb2d10ee81b2b831dc  parallel.go
b88175e65fdcbc01ac08aaf1fd9b5e96  serial.go

程序的主函数调用辅助函数 MD5All，它返回从路径名到摘要值的映射，然后对结果排序并打印：

func main() {
	// Calculate the MD5 sum of all files under the specified directory,
	// then print the results sorted by path name.
	m, err := MD5All(os.Args[1])
	if err != nil {
		fmt.Println(err)
		return
	}
	var paths []string
	for path := range m {
		paths = append(paths, path)
	}
	sort.Strings(paths)
	for _, path := range paths {
		fmt.Printf("%x  %s\n", m[path], path)
	}
}

MD5All 函数是本节讨论的重点。在 serial.go 中，实现不使用任何并发，只是在遍历目录树时逐个读取并计算每个文件的摘要。

// MD5All reads all the files in the file tree rooted at root and returns a map
// from file path to the MD5 sum of the file's contents.  If the directory walk
// fails or any read operation fails, MD5All returns an error.
func MD5All(root string) (map[string][md5.Size]byte, error) {
	m := make(map[string][md5.Size]byte)
	err := filepath.Walk(root, func(path string, info os.FileInfo, err error) error {
		if err != nil {
			return err
		}
		if !info.Mode().IsRegular() {
			return nil
		}
		data, err := ioutil.ReadFile(path)
		if err != nil {
			return err
		}
		m[path] = md5.Sum(data)
		return nil
	})
	if err != nil {
		return nil, err
	}
	return m, nil
}

并行摘要

在 parallel.go 中，我们将 MD5All 拆分为一个两阶段的流水线。第一个阶段 sumFiles 遍历目录树，在新的 goroutine 中对每个文件计算摘要，并通过一个值类型为 result 的 channel 发送结果：

type result struct {
	path string
	sum  [md5.Size]byte
	err  error
}

sumFiles 返回两个 channel：一个用于 results，另一个用于 filepath.Walk 返回的错误。遍历函数会为每个常规文件启动一个新 goroutine 进行处理，然后检查 done。如果 done 已关闭，遍历会立即停止：

func sumFiles(done <-chan struct{}, root string) (<-chan result, <-chan error) {
	// For each regular file, start a goroutine that sums the file and sends
	// the result on c.  Send the result of the walk on errc.
	c := make(chan result)
	errc := make(chan error, 1)
	go func() {
		var wg sync.WaitGroup
		err := filepath.Walk(root, func(path string, info os.FileInfo, err error) error {
			if err != nil {
				return err
			}
			if !info.Mode().IsRegular() {
				return nil
			}
			wg.Add(1)
			go func() {
				data, err := ioutil.ReadFile(path)
				select {
				case c <- result{path, md5.Sum(data), err}:
				case <-done:
				}
				wg.Done()
			}()
			// Abort the walk if done is closed.
			select {
			case <-done:
				return errors.New("walk canceled")
			default:
				return nil
			}
		})
		// Walk has returned, so all calls to wg.Add are done.  Start a
		// goroutine to close c once all the sends are done.
		go func() {
			wg.Wait()
			close(c)
		}()
		// No select needed here, since errc is buffered.
		errc <- err
	}()
	return c, errc
}

MD5All 从 c 接收摘要值。一旦出错，MD5All 会提前返回，并通过 defer 关闭 done：

func MD5All(root string) (map[string][md5.Size]byte, error) {
	// MD5All closes the done channel when it returns; it may do so before
	// receiving all the values from c and errc.
	done := make(chan struct{})
	defer close(done)

	c, errc := sumFiles(done, root)

	m := make(map[string][md5.Size]byte)
	for r := range c {
		if r.err != nil {
			return nil, r.err
		}
		m[r.path] = r.sum
	}
	if err := <-errc; err != nil {
		return nil, err
	}
	return m, nil
}

有界并行

parallel.go 中的 MD5All 实现会为每个文件启动一个新的 goroutine。在包含大量大文件的目录中，这可能分配出比机器可用内存更多的内存。

我们可以通过限制并行读取的文件数量来限制这些内存分配。在 bounded.go 中，我们通过创建固定数量的 goroutines 来读取文件。现在流水线有三个阶段：遍历目录树、读取并计算文件摘要、收集摘要。

第一个阶段 walkFiles 发出目录树中常规文件的路径：

func walkFiles(done <-chan struct{}, root string) (<-chan string, <-chan error) {
	paths := make(chan string)
	errc := make(chan error, 1)
	go func() {
		// Close the paths channel after Walk returns.
		defer close(paths)
		// No select needed for this send, since errc is buffered.
		errc <- filepath.Walk(root, func(path string, info os.FileInfo, err error) error {
			if err != nil {
				return err
			}
			if !info.Mode().IsRegular() {
				return nil
			}
			select {
			case paths <- path:
			case <-done:
				return errors.New("walk canceled")
			}
			return nil
		})
	}()
	return paths, errc
}

中间阶段启动固定数量的 digester goroutines，它们从 paths 接收文件名，并通过 channel c 发送 results：

func digester(done <-chan struct{}, paths <-chan string, c chan<- result) {
	for path := range paths {
		data, err := ioutil.ReadFile(path)
		select {
		case c <- result{path, md5.Sum(data), err}:
		case <-done:
			return
		}
	}
}

与之前的示例不同，digester 不关闭其输出 channel，因为多个 goroutines 会在共享 channel 上发送数据。相反，MD5All 中的代码会在所有 digester 完成后安排关闭该 channel：

// Start a fixed number of goroutines to read and digest files.
c := make(chan result)
var wg sync.WaitGroup
const numDigesters = 20
wg.Add(numDigesters)
for i := 0; i < numDigesters; i++ {
	go func() {
		digester(done, paths, c)
		wg.Done()
	}()
}
go func() {
	wg.Wait()
	close(c)
}()

我们也可以让每个 digester 创建并返回自己的输出 channel，但那样就需要额外的 goroutines 来扇入结果。

最后一个阶段从 c 接收所有 results，然后检查 errc 中的错误。这个检查不能更早进行，因为在此之前，walkFiles 可能阻塞在下游发送值上：

	m := make(map[string][md5.Size]byte)
	for r := range c {
		if r.err != nil {
			return nil, r.err
		}
		m[r.path] = r.sum
	}
	// Check whether the Walk failed.
	if err := <-errc; err != nil {
		return nil, err
	}
	return m, nil
}

结语

本文介绍了在 Go 中构建流式数据流水线的技术。处理此类流水线中的失败颇为棘手，因为流水线中的每个阶段都可能阻塞在向下游发送值上，而下游阶段可能已不再关心传入的数据。我们展示了关闭 channel 如何向流水线启动的所有 goroutines 广播“done”信号，并定义了正确构建流水线的指导原则。

延伸阅读：

• Go Concurrency Patterns（视频）介绍了 Go 并发原语的基础知识及其几种应用方式。
• Advanced Go Concurrency Patterns（视频）涵盖了 Go 原语的更复杂用法，尤其是 select。
• Douglas McIlroy 的论文 Squinting at Power Series 展示了类 Go 的并发如何为复杂计算提供优雅的支持。

原文完。

希望此文能对你有所启发。

延伸阅读

• Go Concurrency Patterns: Pipelines and cancellation 原文：https://go.dev/blog/pipelines
• Go Concurrency Patterns: Context 译文：https://jianghushinian.cn/2026/11/03/go-concurrency-patterns-context/
• Go Concurrency Patterns 演讲：https://go.dev/talks/2012/concurrency.slide
• Advanced Go Concurrency Patterns：https://go.dev/blog/advanced-go-concurrency-patterns
• 本文永久地址：https://jianghushinian.cn/2026/11/04/go-concurrency-patterns-pipelines/

联系我

• 公众号：Go编程世界
• 微信：jianghushinian
• 邮箱：jianghushinian007@outlook.com
• 博客：https://jianghushinian.cn
• GitHub：https://github.com/jianghushinian