簡書不維護(hù)了，歡迎關(guān)注我的知乎：波羅學(xué)的個(gè)人主頁
緊接上文：[翻譯]GO并發(fā)模型一：Pipeline和Cancellation
明確地取消
當(dāng)主函數(shù)在沒有從輸出channel中接收完所有值便退出時(shí)，它必須告訴上游停止數(shù)據(jù)發(fā)送?？梢酝ㄟ^向done channel發(fā)送停止信號(hào)實(shí)現(xiàn)。此處有兩個(gè)可能阻塞的goroutine，所以需發(fā)兩個(gè)值。
func main() {
in := gen(2, 3)
// Distribute the sq work across two goroutine 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)
// Tell the remaining senders we're leaving
done <- struct{}{}
done <- struct{}{}
}
merge中的發(fā)送goroutine用select語句取代了原來的發(fā)送操作，它將負(fù)責(zé)將數(shù)據(jù)發(fā)出和接收done channel的消息。Done將接收的值是空結(jié)構(gòu)體，因?yàn)樵撝禌]有任何意義：它僅僅是用來表明應(yīng)該停止向輸出channel發(fā)送數(shù)據(jù)了。該goroutine將會(huì)不停循環(huán)地從輸入channel中接收數(shù)據(jù)，以確保上游不被阻塞。（待會(huì)我們將會(huì)討論怎么提早從循環(huán)退出）。
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) {
for n := range c {
select {
case out <- n:
case <-done:
}
}
wg.Done()
}
// ... the rest is unchanged ...
}
這種方式的問題是：每個(gè)下游都需知道上游將發(fā)送的數(shù)據(jù)量，以便向其發(fā)送消息實(shí)現(xiàn)提早退出。但毫無疑問，時(shí)刻監(jiān)控已發(fā)送數(shù)量是非?；恼Q，也是非常容易出錯(cuò)的。
我們需要一種在上游goroutine數(shù)量未知或無限大的情況下使其停止的方式。在GO中，我們可以通過關(guān)閉channel來實(shí)現(xiàn)，因?yàn)樵谝殃P(guān)閉的channel上接收數(shù)據(jù)會(huì)被立刻處理并返回一個(gè)零值。
這意味著main函數(shù)中可僅僅通過關(guān)閉done channel來使發(fā)送方解除阻塞。該關(guān)閉操作會(huì)產(chǎn)生一個(gè)有效的廣播信號(hào)并傳遞給發(fā)送方。我們可以擴(kuò)展pipeline中的函數(shù)，使其可以多接受一個(gè)done參數(shù)，然后通過defer語句對(duì)執(zhí)行關(guān)閉以便于在main退出時(shí)發(fā)送給各階段完成信號(hào)來實(shí)現(xiàn)退出。
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 channel關(guān)閉，各個(gè)階段就可以成功返回退出。當(dāng)done被關(guān)閉，merge就會(huì)知道上游會(huì)停止發(fā)送數(shù)據(jù)，merge函數(shù)就會(huì)停止從輸入channel接收數(shù)據(jù)并返回。輸出channel通過defer語句確保所有的wg.Done在函數(shù)時(shí)能被調(diào)用。
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 ...
}
相似地，只要done channel一關(guān)閉，sq函數(shù)也會(huì)立刻返回。通過defer語句，sql函數(shù)確保它們輸出channel一定能被順利關(guān)閉。
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
}
pipeline幾個(gè)指導(dǎo)原則：
- 各個(gè)階段在所有的發(fā)送操作完成便會(huì)關(guān)閉輸出channels；
- 各個(gè)階段會(huì)不停的接收數(shù)據(jù)，直到這些channel都被關(guān)閉或者發(fā)送方不再阻塞；
Pipelines中可以通過為數(shù)據(jù)發(fā)送提供足夠的buffer大小或在接收方確定放棄繼續(xù)接收數(shù)據(jù)時(shí)發(fā)送完成信號(hào)來解除發(fā)送方的阻塞。
對(duì)目錄中的文件執(zhí)行摘要
讓我們來看一個(gè)更實(shí)際的例子.
MD5是一種消息摘要算法，在checksum校驗(yàn)文件方面非常有用。通過命令行工具md5sum，我們打印了一系列文件的摘要值。
% md5sum *.go
d47c2bbc28298ca9befdfbc5d3aa4e65 bounded.go
ee869afd31f83cbb2d10ee81b2b831dc parallel.go
b88175e65fdcbc01ac08aaf1fd9b5e96 serial.go
我們的例子是一個(gè)類似于md5sum的程序，它接受單一目錄作為參數(shù)，并打印該目錄下每個(gè)文件的摘要值。文件是按文件名升序排列打印。
% go run serial.go .
d47c2bbc28298ca9befdfbc5d3aa4e65 bounded.go
ee869afd31f83cbb2d10ee81b2b831dc parallel.go
b88175e65fdcbc01ac08aaf1fd9b5e96 serial.go
程序的main主函數(shù)調(diào)用了一個(gè)名為MD5All的函數(shù)，它返回的是一個(gè)map，key為路徑名，value為摘要值。最后，對(duì)結(jié)果進(jìn)行了排序和打印。
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函數(shù)我們將重點(diǎn)討論。在serial.go文件中，并沒有使用并發(fā)技術(shù)，僅僅是依次讀取每個(gè)文件內(nèi)容并對(duì)其調(diào)用md5.Sum。
// 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
}
并行執(zhí)行摘要
在parellel.go文件中，我們把MD5ALL拆成了兩階段。第一階段，在sumFiles函數(shù)中，它遍歷目錄并在各個(gè)goroutine中執(zhí)行文件摘要，最后將結(jié)果發(fā)送給channel。
type result struct {
path string
sum [md5.Size]byte
err error
}
sumFiles函數(shù)返回了兩個(gè)channels：一個(gè)用于傳輸結(jié)果，另一個(gè)用于返回filepath.Walk的錯(cuò)誤。walk函數(shù)為每個(gè)文件啟動(dòng)了一個(gè)新的goroutine來處理它們，同時(shí)也檢查是否done 。如果done被關(guān)閉，walk函數(shù)將立刻返回。
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函數(shù)從c（channel）中接收摘要值。但發(fā)現(xiàn)錯(cuò)誤，它會(huì)提早返回，并通過defer語句關(guān)閉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
}
對(duì)并行增加限制
parallel.go中的MD5All通過為每個(gè)文件系統(tǒng)一個(gè)新的goroutine實(shí)現(xiàn)。如果一個(gè)目錄中有太多的大文件，這可能會(huì)導(dǎo)致分配的內(nèi)存超過機(jī)器的可用內(nèi)存。
我們可以通過限制并行讀取文件的數(shù)量來限制內(nèi)容分配。在bounded.go文件中，我們創(chuàng)建了固定數(shù)量的goroutines來讀取文件?，F(xiàn)在我們的pipeline涉及了三個(gè)階段：遍歷目錄樹、讀取文件并執(zhí)行摘要和收集摘要結(jié)果。
第一階段，walkFiles，負(fù)責(zé)發(fā)送目錄樹中文件路徑：
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
}
第二階段，我們?yōu)閐igester函數(shù)啟動(dòng)了固定數(shù)量的goroutine，它將從paths中接收文件名處理并發(fā)送摘要結(jié)果給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不會(huì)關(guān)閉輸出channel，因?yàn)樘嗟膅oroutine共用了一個(gè)channel。取而代之，在所有digester執(zhí)行完畢，MD5All會(huì)著手關(guān)閉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)
}()
我們可以為每個(gè)digester創(chuàng)建獨(dú)立的channel，然后返回。但是這樣我們就需要增加額外的goroutines來對(duì)結(jié)果進(jìn)行合并。
最后階段，我們從channel c中接收所有的結(jié)果并檢查errc是否返回了錯(cuò)誤。該檢查無法過早執(zhí)行，因?yàn)檫^早檢查，可能會(huì)導(dǎo)致walkFile阻塞。
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
}
總結(jié)
這篇文章給我們展示了如何在GO中構(gòu)建流式數(shù)據(jù)pipeline。pipeline中處理失敗是需要一定的技巧的，因?yàn)槊總€(gè)嘗試給下游發(fā)送數(shù)據(jù)的階段都可能被阻塞，因?yàn)橄掠慰赡懿辉诮邮丈嫌蔚妮斎霐?shù)據(jù)。我們展示了如何通過關(guān)閉channel來給所有的goroutine發(fā)送 "done" 信號(hào)和定義了正確構(gòu)建pipeline的指導(dǎo)原則。
作者：Sameer Ajmani
