Go 语言并发模型：以并行处理 MD5 为例

简介

Go 语言的并发原语允许开发者以类似于 Unix Pipe 的方式构建数据流水线 (data pipelines)，数据流水线能够高效地利用 I/O 和多核 CPU 的优势。

本文要讲的就是一些使用流水线的一些例子，流水线的错误处理也是本文的重点。

阅读建议

本文是"Go 语言并发模型：像 Unix Pipe 那样使用 channel" 一文的下半部分，但重点在于实践。如果你对 channel 已经比较熟悉，则可以独立阅读。 如果你对 channel 和 go 两个关键字不太熟悉，建议先阅读上半部分。

本文所使用的例子是批量计算文件的 MD5 值，实现了 linux 下的 md5sum 命令。 我们首先会讲到 md5sum 的单线程版本，逐步深入到并发的初级和高级版本。

本文中绝大多数讲解都是基于代码进行的。在文章末尾"相关链接"中可以下载三个版本的 md5sum 的实现。

单线程版的 md5sum

MD5 是一种广泛用于文件校验的 hash 算法。 Linux 下的 md5sum 命令会打印一组文件的 md5 值。它的使用方式如下：

% md5sum *.go c33237079343a4d567a2a29df0b8e46e bounded.go a7e3771f2ed58d4b34a73566d93ce63a parallel.go 1dc687202696d650594aaac56d579179 serial.go 

我们的示例程序类似于 md5sum ，但是它接收文件夹作为参数，并打印出每个文件的 md5 值，打印结果按照路径排序。 下面这个例子是 打印当前目录下所有文件的 md5 值：

% go run serial.go . c33237079343a4d567a2a29df0b8e46e bounded.go a7e3771f2ed58d4b34a73566d93ce63a parallel.go 1dc687202696d650594aaac56d579179 serial.go 

程序的 main 函数调用辅助函数 MD5All ，它会返回路径名称到 md5 值的一个映射。 main 函数中对结果进行排序以后，打印出来：

func main() { // 计算特定目录下所有文件的 md5 值， // 然后按照路径名顺序打印结果 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的实现中，我们没有使用并发，而是逐个读取和计算 filepath.Walk 生成的目录和文件。代码如下：

// MD5All 读取 root 目录下的所有文件，返回一个 map // 该 map 存储了 文件路径到文件内容 md5 值的映射 // 如果 Walk 执行失败，或者 ioutil.ReadFile 读取失败， // MD5All 都会返回错误 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 } 

上面的代码中，filepath.Walk 接收两个参数，文件路径和函数指针。 只要是函数签名和返回值 满足 func(string, os.FileInfo, error) error，均可以作为第二参数传递给 filepath.Walk 。

点击 serial.go 下载单线程版本的 md5sum 。

并发版的 md5sum

点击 parallel.go 下载并发版 md5sum 的代码。

在这个版本的实现中，我们把 MD5All 切割成两个阶段的流水线。 第一阶段是 sumFiles ，它遍历文件树，每个文件都在一个新的 goroutine 里计算 md5 值，然后将结果发送到一个 result 类型的 channel 里。 result 类型的定义如下：

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

sumFiles 返回两个 channel ，一个用于接收 md5 计算的结果，一个用于接收 filepath.Walk 产生的错误。 Walk 函数为每一个文件创建一个 goroutine ，然后检查 done channel 。如果 done channel 被关闭， walk 函数立即停止执行。代码示例如下：

func sumFiles(done <-chan struct{}, root string) (<-chan result, <-chan error) { // 对于每一个普通文件，启动一个 gorotuine 计算文件 md5 值， // 然后 将结果发送到 c 。 // walk 的错误结果发送到 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() }() // done channel 关闭时，终止 walk 函数 select { case <-done: return errors.New("walk canceled") default: return nil } }) // Walk 函数已经返回，所以 所有对 wg.Add 的调用都会结束 // 启动一个 goroutine ， 它会在所有发送都结束时，关闭 c 。 go func() { wg.Wait() close(c) }() // 这里不需要 select 语句，应为 errc 是缓冲管道 errc <- err }() return c, errc } 

MD5All 从 c 接收 md5 值。 MD5All 遇到错误时会提前返回，通过 defer 语句关闭 done channel ：

func MD5All(root string) (map[string][md5.Size]byte, error) { // MD5All 在函数返回时关闭 done channel // 在从 c 和 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 } 

限制并发量

在 并发版 MD5All (parallel.go) 的实现中， 我们为每个文件创建了一个 goroutine 。如果一个目录中包含很多大文件，可能出现 OOM 。

我们对并发读取的文件数目稍作限制，进而限制内存的分配。点击 bounded.go 查看限制并发版本的 md5sum 。 为了实现限制的目的，我们创建固定数量的 goroutine 用于读取文件。 这里的流水线包含三个阶段：遍历文件和目录、读取并计算 md5 值、搜集和整合计算结果。

第一阶段时 walkFiles ，它生成一个目录下每个普通文件的路径。代码如下：

func walkFiles(done <-chan struct{}, root string) (<-chan string, <-chan error) { paths := make(chan string) errc := make(chan error, 1) go func() { // Walk 函数返回时，关闭 channel paths defer close(paths) // 这里不需要 select ，因为 errc 是缓冲 channel 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 } 

第二阶段创建固定个数的 goroutine digester ，每个 digester 从 paths channel 读取文件名，并将结果发送给 c 。代码如下：

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 c ，因为 多个 digester 在共享这个 channel 。 关闭操作放到 MD5All 中实现，当所有 digester 运行结束时， MD5All 关闭这个 channel 。代码如下：

 // 启动固定数量的 goroutine 处理文件 c := make(chan result) var wg sync.WaitGroup const numDigesters = 20 wg.Add(numDigesters) for i := 0; i < numDigestes; i++ { go func() { digester(done, paths, c) wg.Done() }() } go func() { wg.Wait() close(c) }() 

我们可以让每个 digester 创建和返回自己的输出 channel 。如果这样做，我们还需要额外的 goroutine 去合并结果。

第三阶段从 channel c 接收结果，并从 channel errc 读取错误信息并执行检查。 检查操作不能在 c 读取结束之前完成，因为 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 语言并发模型，使用 Go 内置的 channel 类型和 go 关键字实现高并发和并发控制的主题就先到这里。 在最近发布的 go 1.7 中，在核心库中广泛加入了对 context 的支持，以便更好地控制并发和超时。但在这之前 golang.org/x/net/context 包就一直存在，下一期我们将对 context 包及其应用场景进行讨论。

相关链接：

1. 原文链接
2. serial.go
3. parallel.go
4. bounded.go
5. golang.org/x/net/context

扫码关注微信公众号“深入 Go 语言”