go - 尝试使用 channel 获取 FIFO 处理管道

Question

我正在尝试使用一个简单的假示例在 Go 中编写顺序处理管道。它遍历一个假目录并运行一些转换。因此，它们之间共享一个字符串 channel 。在一个函数写入数据后，第二个函数读取它。

在我看来，当我在 WalkFakeDirectory 函数前面放置一个 go 关键字时，它只起作用并且也按顺序起作用，如下面的代码示例 (playground) 所示。

如果有人能解释一下这是如何工作的，我们将不胜感激？

package main 
import (
"fmt"
"strings"
"sync"
"time"
)

func main() {

done := make(chan int)
path := make(chan string)

defer close(done)

//var wg sync.WaitGroup - Not working too
//wg.Add(1)

fmt.Println("walking file path")

go WalkFakeDirectoy(done, path)

//wg.Add(1)

ConvertToUpperCase(path, done)

//wg.Wait()

fmt.Println("done!")
//time.Sleep(2000) // Not working 
}

func ConvertToUpperCase(files chan string, done chan int) {

   for file := range files {
      fmt.Println("converting data data", strings.ToUpper(file))
   }
}

func WalkFakeDirectoy(done chan int, path chan<- string) {

   func() {
    list := []string{"abc", "def", "fgh", "ijk", "mnn"}

    for _, file := range list {
        fmt.Println("getting data", file)
        path <- file
        time.Sleep(3000)
      }
   }()
}

Answer 1

This Go blog post关于管道应该有足够的信息来构建你自己的。它的要点是一个代码示例:

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

func sq(in <-chan int) <-chan int {
    out := make(chan int)
    go func() {
        for n := range in {
            out <- n * n
        }
        close(out)
    }()
    return out
}
func sample_pipeline() {
    // Set up the pipeline.
    c := gen(2, 3)
    out := sq(c)
    out2 := sq(out)

    // Consume the output.
    for o := range out2 {
        fmt.Println(o)
    }
}

func main() {
    sample_pipeline()
}

sq 是一个流水线阶段 - 它接受一个带输入的 channel 并返回一个带输出的 channel (输入的平方值)。 sample_pipeline 设置一个两级管道并将具有两个值的生成器连接到它。

重要的是要注意完成是如何完成的——每个管道阶段都是一个 goroutine 来完成管道阶段的工作(等待来自输入管道的新数据，处理它，发送输出)。当每个阶段的输入管道完成时( channel 上的范围循环停止)，它关闭自己的 channel 。关闭 channel 是 Go 中发出“此 channel 已完成”信号的规范方式。