go - 如何在管道中使用 sync.WaitGroup

Question

我正在尝试在 Go 中实现管道，并且存在程序在其余 goroutine 完成之前退出主 goroutine 的问题。

请使用等待组帮助解决此问题。

package main

import (
    "fmt"
    "sync"
)

var wg sync.WaitGroup

func main() {
    c1 := make(chan string)
    c2 := make(chan string)

    go sender(c1)
    go removeDuplicates(c1, c2)
    go printer(c2)

    wg.Wait()
}

func sender(outputStream chan string) {
    wg.Add(1)
    wg.Done()

    for _, v := range []string{"one", "one", "two", "two", "three", "three"} {
        outputStream <- v
    }

    close(outputStream)
}

func removeDuplicates(inputStream, outputStream chan string) {
    wg.Add(1)
    wg.Done()

    temp := ""

    for v := range inputStream {
        if v != temp {
            outputStream <- v
            temp = v
        }
    }

    close(outputStream)
}

func printer(inputStream chan string) {
    wg.Add(1)
    wg.Done()

    for v := range inputStream {
        fmt.Println(v)
    }
}

当我在这种情况下使用 time.Sleep 时，程序成功运行。

Answer 1

使用等待组的第一条规则：不要Add()从你要等待的 goroutine 中调用。有可能，这Wait()将被调用before Add()，这不是您所期望的。第二条规则：Done()最后调用，而不是开始调用。因此，解决方法是：

func main() {
    c1 := make(chan string)
    c2 := make(chan string)

    wg.Add(3)

    go sender(c1)
    go removeDuplicates(c1, c2)
    go printer(c2)

    wg.Wait()
}

func sender(outputStream chan string) {
    defer wg.Done()

    for _, v := range []string{"one", "one", "two", "two", "three", "three"} {
        outputStream <- v
    }

    close(outputStream)
}

// etc for other two functions