通道过早终止

Question

我正在为每个执行转换的管道制作一系列 go 例程的原型。例程在所有数据通过之前终止。

我查看了 Donavan 和 Kernighan 的书，并在 Google 上搜索了解决方案。

这是我的代码：

package main

import (
    "fmt"
    "sync"
)

func main() {
    a1 := []string{"apple", "apricot"}

    chan1 := make(chan string)
    chan2 := make(chan string)
    chan3 := make(chan string)

    var wg sync.WaitGroup

    go Pipe1(chan2, chan1, &wg)
    go Pipe2(chan3, chan2, &wg)
    go Pipe3(chan3, &wg)

    func (data []string) {
        defer wg.Done()
        for _, s := range data {
            wg.Add(1)
            chan1 <- s
        }
        go func() {
            wg.Wait()
            close(chan1)
        }()
    }(a1)
}

func Pipe1(out chan<- string, in <-chan string, wg *sync.WaitGroup) {
    defer wg.Done()
    for s := range in {
        wg.Add(1)
        out <- s + "s are"
    }
}
func Pipe2(out chan<- string, in <-chan string, wg *sync.WaitGroup) {
    defer wg.Done()
    for s := range in {
        wg.Add(1)
        out <- s + " good for you"
    }
}
func Pipe3(in <-chan string, wg *sync.WaitGroup) {
    defer wg.Done()
    for s := range in {
        wg.Add(1)
        fmt.Println(s)
    }
}

我的预期输出是：

apples are good for you
apricots are good for you

运行main的结果不一致。有时我得到两条线。有时我只是得到苹果。有时什么也没有输出。

Answer 1

正如 Adrian 已经指出的，您的 WaitGroup.Add 和 WaitGroup.Done 调用不匹配。然而，在这种情况下，“我完成了”信号通常是通过关闭输出通道来给出的。仅当多个 goroutine 之间共享工作时才需要 WaitGroups（即多个 goroutine 使用相同的通道），这里不是这种情况。

package main

import (
    "fmt"
)

func main() {
    a1 := []string{"apple", "apricot"}

    chan1 := make(chan string)
    chan2 := make(chan string)
    chan3 := make(chan string)

    go func() {
        for _, s := range a1 {
            chan1 <- s
        }

        close(chan1)
    }()

    go Pipe1(chan2, chan1)
    go Pipe2(chan3, chan2)

    // This range loop terminates when chan3 is closed, which Pipe2 does after
    // chan2 is closed, which Pipe1 does after chan1 is closed, which the
    // anonymous goroutine above does after it sent all values.
    for s := range chan3 {
        fmt.Println(s)
    }
}

func Pipe1(out chan<- string, in <-chan string) {
    for s := range in {
        out <- s + "s are"
    }

    close(out) // let caller know that we're done
}

func Pipe2(out chan<- string, in <-chan string) {
    for s := range in {
        out <- s + " good for you"
    }

    close(out) // let caller know that we're done
}

在操场上试试看：https://play.golang.org/p/d2J4APjs_lL

Answer 2

您在 goroutine 中调用 wg.Wait，因此允许 main 在其他例程完成之前返回（因此您的程序退出）。这会导致您看到的行为，但仅退出 goroutine 是不够的。

一般来说，您还误用了WaitGroup；你的Add 和Done 调用彼此不相关，而且你的Dones 没有Adds 那么多，所以WaitGroup 永远不会结束。如果您在循环中调用Add，那么每次循环迭代也必须导致Done 调用；正如您现在所拥有的，您在每个循环之前defer wg.Done()，然后在循环内调用Add，从而产生一个Done 和许多Adds。此代码需要进行重大修改才能按预期工作。