尝试使用通道获取 FIFO 处理管道答案

【问题标题】：Trying to get FIFO processing pipeline using channel尝试使用通道获取 FIFO 处理管道
【发布时间】：2019-06-14 15:30:41
【问题描述】：

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

在我看来，当我在 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)
      }
   }()
}

【问题讨论】：

您正在使用无缓冲通道作为路径。因此，如果您调用 WalkFakeDirectory，它会在将第一个文件插入频道后卡住，因为直到此时没有人在听。但是，一旦您在 go 例程中添加 run walkFakeDirectory，ConvertToUpperCase 就会监听通道，并继续使用通道中的数据。希望这能让你的疑惑清楚
处理完 WalkFakeDirectory 中的所有文件后，您需要 close(path)。整个 done 或 WaitGroup 的东西是完全没有必要的。
@sourabh1024 我认为您是对的，并提供了我正在寻找的答案。谢谢。

标签： go concurrency channel

【解决方案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 是一个流水线阶段——它接受一个带有输入的通道并返回一个带有输出的通道（输入的平方值）。 sample_pipeline 设置了一个两级管道，并连接了一个具有两个值的生成器。

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

【讨论】：