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 < numDigesters; 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 包及其应用场景进行讨论。

原文发布于微信公众号 - Golang语言社区(Golangweb)

原文发表时间:2017-01-09

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

发表于

我来说两句

0 条评论
登录 后参与评论

相关文章

来自专栏用户2442861的专栏

JavaWeb工程中web.xml基本配置

        先说下我记得xml规则,必须有且只有一个根节点,大小写敏感,标签不嵌套,必须配对。

33910
来自专栏PhpZendo

Nginx 烹调书

本书是「Complete Nginx Cookbook」一书的部分中英文对照翻译版本。

16010
来自专栏小樱的经验随笔

【批处理学习笔记】第九课:批处理符号(2)

四、| 这是一个管道传输命令,意思是将上一命令执行的结果传到下一个命令去处理 例如: dir c:\|find "txt" 以上命令是:查找C:\所有,...

34170
来自专栏运维小白

5.5 进入编辑模式

编辑模式:就是进入到可以编辑文本文件的模式 ? 进入编辑模式方法: i ,直接在当前光标处进入编辑模式 I (大写I),光标直接移动到该行的行首,并进入编辑...

21270
来自专栏GreenLeaves

WCF系列教程之WCF服务协定

本文参考自:http://www.cnblogs.com/wangweimutou/p/4422883.html,纯属读书笔记,加深记忆 一、服务协定简介: 1...

19370
来自专栏水击三千

UML学习-时序图

时序图(Sequence Diagram)是显示对象之间交互的图,这些对象是按时间顺序排列的。顺序图中显示的是参与交互的对象及其对象之间消息交互的顺序。时序图...

41790
来自专栏一个爱瞎折腾的程序猿

常用cmd代码片段及.net core打包脚本分享

保存:set currentPath=%cd% 输出:echo %currentPath

13930
来自专栏运维小白

10.7 free命令

监控系统状态 free 查看内存使用情况 free -m / -g / -h buffer/cache区别 公式:total=used+free+buff/ca...

27870
来自专栏玄魂工作室

看代码学渗透 Day5 - escapeshellarg与escapeshellcmd使用不当

--------------------------------------------------------------------------------...

41820
来自专栏Django中文社区

博客文章详情页

首页展示的是所有文章的列表,当用户看到感兴趣的文章时,他点击文章的标题或者继续阅读的按钮,应该跳转到文章的详情页面来阅读文章的详细内容。现在让我们来开发博客的详...

46170

扫码关注云+社区

领取腾讯云代金券