Go并发模型:流水线与取消(Pipelines and cancellation译文)
Posted -_-void
tags:
篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了Go并发模型:流水线与取消(Pipelines and cancellation译文)相关的知识,希望对你有一定的参考价值。
Go并发模型:流水线与取消 (Go Concurrency Patterns: Pipelines and cancellation)
本文不只是简单的翻译,有些地方根据自己的理解用中文习惯重新组织了语言,所以可能会有局部的顺序不同,但是读起来更通顺。所以如果对文中任何部分有疑问可以直接体温,保证知无不言。
英文原版: https://blog.golang.org/pipelines
简介
go语言的并发机制可以使CPU及IO更高效的处理数据流。本文展示几个例子来介绍下流水线以及执行操作失败时的细节,还有处理异常时所用的技术。
流水线是啥
在go里面并没有对流水线的正式定义,它就是各种并发程序。通俗来说,一个流水线就是通过channel连接的一组stage,每个stage就是运行着同一function的一组goroutine。这些goroutine完成如下任务
- 从已绑定的输入channel中读取上游数据
- 处理读到的数据,通常会产生新的数据
- 将新的数据发送到已绑定的输出channel
第一个stage只有输入channel,最后一个stage只有输出channel,其他每个stage都有若干个输入输出channel。第一个stage有时被称为srouce
或者producer
,称最后一个stage为sink
或consumer
下面从一个简单的例子开始,之后还有其他相关的例子。
数字做平方
假设有一个包含三个stage的流水线。
第一个stage是一个叫gen的function,它把参数中传进来的数字list在goroutine中放进channel并返回这个channel,在所有的数字发完之后关闭这个channel:
func gen(nums ...int) <-chan int
out := make(chan int)
go func()
for _, n := range nums
out <- n
close(out)
()
return out
第二个stage是个叫sq的function,它从一个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方法,建立起流水线并充当第三个stage。即从channel中读取数字并挨个print:
func main()
// 建立流水线
c := gen(2, 3)
out := sq(c)
// print结果
fmt.Println(<-out)
fmt.Println(<-out)
由于sq的输入跟输出是同类型的channel,所以我们可以让它嵌套一下。当然这个main也可以写成上面两个stage风格:
func main()
for n := range sq(sq(gen(2, 3)))
fmt.Println(n)
扇入(fan-in),扇出(fan-out)
扇入指的是在一个function中处理多个输入channel并将结果输出到一个输出channel,并在所有输入channel关闭后关闭输出channel。扇出就是一个channel可以作为多个function的输入channel,这就相当于有多个worker分发处理同一组任务。现在我们来调整一下流水线,让平方操作分发给两个sq实例,当然所以sq实例共享同一个输入channel。然后我们还需要一个新的function来扇入这些数据,就叫merge吧。首先调整一下main:
func main()
in := gen(2, 3)
c1 := sq(in)
c2 := sq(in)
for n := range merge(c1, c2)
fmt.Println(n)
merge方法中给每个输入channel起个goroutine来读数据,然后放进同一个输出channel,另外还需要一个goroutine等所有的输入channel关闭后关闭输出channel
func merge(cs ...<-chan int) <-chan int
// 用来等待输入channel的wg
var wg sync.WaitGroup
// 唯一的输出channel
out := make(chan int)
// 将处理输入的func定义成变量
output := func(c <-chan int)
for n := range c
out <- n
wg.Done()
// 等待数跟输入channel数一样
wg.Add(len(cs))
// 为每个输入channel开goroutine
for _, c := range cs
go output(c)
// 开goroutine进行等待与关闭输出channel
go func()
wg.Wait()
close(out)
()
// 返回输出channel
return out
急停(stoping short)
上面离职中的流水线有如下两个模式:
- stage执行完所有的发送操作后关闭输出channel
- stage在输入channel关闭前持续读取数据
这种模式允许每个stage在接收数据时可以用range循环,并且保证每个goroutine在将所有的数据成功发送给下游之后立即退出。但是实际用到的流水线,stage不一定总是接收所有的输入数据,这跟设计有关。例如有时候stage只需要输入数据的一部分子集就可以完成任务。很多时候stage还会提前退出,比如上游的stage传进来一个error。还有中情况是stage已经不需要再接收数据了,并且这种情况下还希望上游stage不在继续产生下游不需要的数据。在上面的例子中,如果有个stage出错不能在消费输入channel的数据,那么这个channel的发送端将会永久阻塞:
// 消费中的第一条数据
out := merge(c1, c2)
fmt.Println(<-out)
return
// 从此不再消费第二条及之后的数据,这样想往out发数据的地方就会阻塞
这显然会导致协程泄露,goroutine使用的cpu等运行时资源,还有它自己的堆栈里的数据也不会被GC回收。注意goroutine是没有回收机制的,他们只能自己退出。如果pipeline中的下游stage不能再继续消费数据,那么我们需要让上游的stage知道并作出响应(退出或其他操作)。我们可以给channel加缓冲区,当缓冲区还有空间的时候向channel发送数据就不会阻塞:
c := make(chan int, 2)
c <- 1 // 直接发送
c <- 2 // 直接发送
c <- 3 // 如果channel另一头没消费前面的数据,这里会一直阻塞,直到缓冲区有空位
如果可以事先直到要处理的数据量,那就可以直接用带缓冲区的channel来简化代码。拿上面的例子来说,现在可以直接重写gen方法,有了缓冲区甚至可以不开新的goroutine了:
func gen(nums ...int) <-chan int
out := make(chan int, len(nums))
for _, n := range nums
out <- n
close(out)
return out
同样需要给在merge中返回的channel加个缓冲区避免被下游阻塞:
func merge(cs ...<-chan int) <-chan int
// 只展示被修改这行代码,其他地方不动
//out := make(chan int)
out := make(chan int, 1) // 在这里写个足够大的值来存放没有被消费的数据
虽然这样做“解决了”goroutine阻塞的问题,但这其实不是正解。这里设置的缓冲区的大小“1”依赖于已知merge方法能接到多少数据以及下游的stage能消费多少。这在健壮性上差点意思,如果给gen加了点可选数据,或者说下游需要的数据变少时,仍然会有goroutine阻塞。所以我们需要找到一种方法让下游的stage可以把自己准备停止消费数据的消息“告诉”上游。
明确取消(explicit cancellation)
当main决定不接受输入channel的剩余数据时,它必须通知上游的stage来丢弃还没发送的数据。可以通过加一个channel来实现,可以称它为done,由于上游有两个发送方,所以需要给done发两个数据:
func main()
in := gen(2,3)
c1 := sq(in)
c2 := sq(in)
done := make(chan interface, 2)
out := merge(done, c1, c2)
fmt.Println(<-out)
// 通知发送方
done <- struct
done <- struct
然后他的发送方需要用select来调整一下逻辑,select中用两个case分别对应向out发数据以及从done收数据。done中的数据类型为空struct是因为其值不需要被关心,起作用的是有或者没有。如下修改之后,这个output所在的goroutine就可以继续for循环,就不会阻塞它的上游stage了。(后面会继续说怎么让这个循环早点退出,毕竟下游都不要数据了,这里空跑也没意义)
func merge(done <-chan struct, cs ...<-chan int) <-chan int
// 其他代码不变,这里只写output这段
output := func(c <-chan int)
for n := range c
select
case out <- n:
case <-done:
wg.Done()
这种解决办法有个弊端:每个下游的接收方都需要知道会被它阻塞的上游发送方的数量,以及挨个给他们发送结束的信号,持续维护这个数目比较蛋疼而且容易出错。因此需要一种方式来告知不可预知的goroutine来停止发送数据,在go里可以通过关闭channel来实现。因为从已关闭的channel读数据会立即读到一个该类型的零值数据。这就意味着main可以通过关闭done通道来非阻塞的通知所有上游,这样一来关闭操作就成了非常高效的广播。现在可以扩展一下流水线里的每个function,让他们多接受一个done参数,然后在main中通过defer来调用close,这样main无论在什么情况下退出都会向上游stage发送退出信号。
func main()
// 上面说的比较清楚了,这里不放注释了
done := make(chan struct)
defer close(done)
in := gen(done, 2, 3)
c1 := sq(done, in)
c2 := sq(doen, in)
out := merge(done, c1, c2)
fmt.Println(<-out)
然后再调整一下merge,这次放上完整的代码
func merge(cs ...<-chan int) <-chan int
// 用来等待输入channel的wg
var wg sync.WaitGroup
// 唯一的输出channel
//out := make(chan int)
out := make(chan int, 1) // 在这里写个足够大的值来存放没有被消费的数据
// 将处理输入的func定义成变量
output := func(c <-chan int)
// 本次改动:done以defer形式调用,放在函数开头
defer wg.Done()
for n := range c
select
case out <- n:
case <-done:
// 本次改动:加上return,done之后直接退出
return
// 等待数跟输入channel数一样
wg.Add(len(cs))
// 为每个输入channel开goroutine
for _, c := range cs
go output(c)
// 开goroutine进行等待与关闭输出channel
go func()
wg.Wait()
close(out)
()
// 返回输出channel
return out
上面代码中做了“本次改动”之后就可以在收到done消息时结束方法,并调用waitGroup的done方法。现在整个流水线中的每个stage都是独立的自由退出的(即收到done被关闭消息时直接退出)。同时merge中的每个output所在goroutine,可以在收到close消息时不管上游数据是否还有剩余就直接退出。
类似的sq方法也可以在收到close消息时立即退出,并且可以通过defer来保证关闭输出channel:
func sq(done <-chan struct, int <-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
关于流水线有这样两条规则:
- stage在所有的数据发送完成之后,关闭输出channel
- stage应该持续从输入channel接收数据,除非这些channel关闭或者发送方是非阻塞式的。(非阻塞式的发送方:比如发了10条数据,channel的缓冲区也为10)
总的来说,流水线中让发送方stage不阻塞的方式有两种,一个是让channel有足够的buffer,另一个就是接收方显示通知发送方不需要继续发数据了。
多路处理
再来看一个更实际的流水线。
MD5是是一种摘要算法,文件校验和通常用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
main方法中用到的是MD5All
这个方法,它可以返回一个map,文件名作为key,摘要作为value:
func main()
// 计算路径下所有文件的sum,然后按文件名排序输出
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)
先来看下serial.go
中的MD5All
,这里是没用到并发的,只是简单的挨个读文件然后计算:
// MD5All读取root下所有文件,返回一个map,map的key是文件名,value是文件的md5。如果发生错误会返回一个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 nik, err
return m, nil
并行处理
而在parallel.go
文件中将MD5All
拆分成了包含俩stage的流水线。第一个stage叫sumFiles
,用来遍历并为每个文件开新的goroutine来计算摘要,然后将结果发给输出channel。先看下这个result的定义:
type result struct
path string
sum [md5.Size]byte
err error
sumFiles
返回两个channel,一个用来传递result
,另一个传递filepath.Walk
产生的error。walk里面会给每个文件开新的goroutine来读文件与计算摘要,然后检查done
这个channel。如果done
关闭了,walk会立即停止:
func sumFiles(done <-chan struct, root string) (<-chan result, <-chan error)
// 给每个常规文件开独立的goroutine来处理,之后把result发送到c这个channel
// walk产生的error发送到errc这个channel
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 <- resultpath, md5.Sum(data), err:
case <-done:
wg.Done()
()
// 如果done已经close了,终止walk这个方法
select
case <-done:
return errors.New("walk canceled")
delfault:
return nil
)
// 代码运行到这,说明walk方法已经return了,也就意味着所有的`wg.Add(1)`都执行了。因此在这里开一个goroutine来执行等待,就可以保证在所有的`wg.Done()`执行后再关闭c这个channel
go func()
wg.Wait()
close(c)
()
// 这里不用select,因为errc定义时就给了大小为1的缓冲区
errc <- err
()
return c, errc
MD5All
从c这个channel中接收摘要结果,如果有error会提前return。done这个channel的关闭时通过defer的方式执行的:
func MD5All(root string) (map[string][md5.Size]byte, error)
// done会在方法return时关闭,方法返回时可能并没有处理完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 nik, r.err
m[r.path] = r.sum
if err := <-errc; err != nil
return nil, err
return m, nil
有限并行(Bounded parallelism)
上面带并行的MD5All
方法给每个文件都开goroutine去处理,但是如果这个文件中有很多大文件的话,这可能会比较耗内存。所以需要限制并行处理的文件数量。在bounded.go
中设置了一个变量来作为goroutine数最大值,这样一来流水线就成了3个stage:遍历文件、读文件与计算摘要、汇总摘要。第一个stage负责产出文件path:
func walkFiles(done <-chan struct, root string) (<-chan string, <-chan error)
pahts := make(chan string)
errc := make(chan error, 1)
go func()
// walk结束时关闭paths这个channel
defer close(paths)
// errc带了缓冲区,所以这里不需要用select
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
第二个stage是预设数量的goroutine来从paths通道读取文件路径,处理之后将result从c通道发送出去,这个stage可以命名为digester
:
func digester(done <-chan struct, paths <-chan string, c <-chan result)
for paths := range paths
data, err := iotuil.ReadFile(path)
select
case c <- resultpath, md5.Sum(data), err:
case <-done:
return
digester
跟之前的例子不一样,它不会主动关闭输出channel。这个channel是多个goroutine在同时向其发送数据的。所以它是在MD5All
中将其中数据消费之后再关闭的:
// 被`digester`们共享的输出channel
c := make(chan result)
// 用来等待所有的`digester`完成
var wg sync.WaitGroup
// 用来指定`digester`的最大数量
const numDigesters = 20
wg.Add(numDigesters)
// 启动指定数量的`digester`
for i := 0; i < numDigesters; i++
go func()
digester(done, paths, c)
wg.Done()
()
// 在所有digester完成之后关闭共享通道
go func()
wg.Wait()
close(c)
实际上也可以让每个digester
创建独立的输出channel,但这样的话就需要再加一个goroutine来扇入(fan-in)这些输出channel了。最后一个stage就是从c这个共享channel中读取所有的result,以及检查errc里的error。这个检查操作不会提前执行,因为这样的话walkFiles
可能会阻塞下游的stage。
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
总结
本文主要介绍了在go语言中构建数据流水线,在流水线中每个stage处理异常可能会阻塞下游的stage,下游的stage也有可能不需要后续的上游数据。文中介绍了通过关闭done
通道来向所有stage“广播”消息以及正确的定义流水线。
以上是关于Go并发模型:流水线与取消(Pipelines and cancellation译文)的主要内容,如果未能解决你的问题,请参考以下文章