sync.WaitGroup 在 Channels 上的优势是什么？

Question

我正在开发一个并发 Go 库，我偶然发现了 goroutines 之间的两种不同的同步模式，它们的结果相似：package mainimport (    "fmt"    "sync"    "time")var wg sync.WaitGroupfunc main() {    words := []string{"foo", "bar", "baz"}    for _, word := range words {        wg.Add(1)        go func(word string) {            time.Sleep(1 * time.Second)            defer wg.Done()            fmt.Println(word)        }(word)    }    // do concurrent things here    // blocks/waits for waitgroup    wg.Wait()}package mainimport (    "fmt"    "time")func main() {    words := []string{"foo", "bar", "baz"}    done := make(chan bool)    // defer close(done)    for _, word := range words {        // fmt.Println(len(done), cap(done))        go func(word string) {            time.Sleep(1 * time.Second)            fmt.Println(word)            done <- true        }(word)    }    // Do concurrent things here    // This blocks and waits for signal from channel    for range words {        <-done    }}我被告知它的sync.WaitGroup性能稍好一些，而且我已经看到它被普遍使用。但是，我发现频道更惯用。使用sync.WaitGroupover channel的真正优势是什么和/或当它更好时可能会是什么情况？

Answer 1

独立于第二个示例的正确性（如评论中所述，您没有按照自己的想法行事，但很容易修复），我倾向于认为第一个示例更容易掌握。

现在，我什至不会说频道更惯用了。通道是 Go 语言的一个标志性特性，并不意味着尽可能地使用它们是惯用的。Go 中的惯用语是使用最简单、最容易理解的解决方案：在这里，WaitGroup传递含义（您的主要功能是Wait让工人完成）和机制（工人通知他们何时完成Done）。

除非您处于非常特殊的情况，否则我不建议在此处使用通道解决方案。

Answer 2

这取决于用例。如果您要分派一次性作业以并行运行而无需知道每个作业的结果，那么您可以使用WaitGroup. 但是如果你需要从 goroutines 收集结果，那么你应该使用一个通道。

由于通道双向工作，我几乎总是使用通道。

另一方面，正如评论中所指出的，您的频道示例未正确实现。您需要一个单独的通道来指示没有更多的工作要做（一个例子是这里）。在您的情况下，由于您事先知道字数，因此您可以只使用一个缓冲通道并接收固定次数以避免声明关闭通道。

Answer 3

对于您的简单示例（表示作业完成），这WaitGroup是显而易见的选择。Go 编译器非常友好，不会怪你使用通道来发送简单的完成任务信号，但一些代码审查者会这样做。

"WaitGroup 等待一组 goroutines 完成。主 goroutine 调用Add(n)设置要等待的 goroutines 的数量。然后每个 goroutines 运行并Done()在完成时调用。同时，Wait 可用于阻塞直到所有goroutine 已经完成了。”

words := []string{"foo", "bar", "baz"}

var wg sync.WaitGroup

for _, word := range words {

    wg.Add(1)

    go func(word string) {

        defer wg.Done()

        time.Sleep(100 * time.Millisecond) // a job

        fmt.Println(word)

    }(word)

}

wg.Wait()

可能性仅受您的想象力的限制：

通道可以被缓冲：

words := []string{"foo", "bar", "baz"}

done := make(chan struct{}, len(words))

for _, word := range words {

    go func(word string) {

        time.Sleep(100 * time.Millisecond) // a job

        fmt.Println(word)

        done <- struct{}{} // not blocking

    }(word)

}

for range words {

    <-done

}

通道可以是无缓冲的，你可以只使用一个信号通道（例如chan struct{}）：

words := []string{"foo", "bar", "baz"}

done := make(chan struct{})

for _, word := range words {

    go func(word string) {

        time.Sleep(100 * time.Millisecond) // a job

        fmt.Println(word)

        done <- struct{}{} // blocking

    }(word)

}

for range words {

    <-done

}

您可以限制具有缓冲通道容量的并发作业数：

t0 := time.Now()

var wg sync.WaitGroup

words := []string{"foo", "bar", "baz"}

done := make(chan struct{}, 1) // set the number of concurrent job here

for _, word := range words {

    wg.Add(1)

    go func(word string) {

        done <- struct{}{}

        time.Sleep(100 * time.Millisecond) // job

        fmt.Println(word, time.Since(t0))

        <-done

        wg.Done()

    }(word)

}

wg.Wait()

您可以使用频道发送消息：

done := make(chan string)

go func() {

    for _, word := range []string{"foo", "bar", "baz"} {

        done <- word

    }

    close(done)

}()

for word := range done {

    fmt.Println(word)

}

基准：

    go test -benchmem -bench . -args -n 0

# BenchmarkEvenWaitgroup-8  1827517   652 ns/op    0 B/op  0 allocs/op

# BenchmarkEvenChannel-8    1000000  2373 ns/op  520 B/op  1 allocs/op

    go test -benchmem -bench .

# BenchmarkEvenWaitgroup-8  1770260   678 ns/op    0 B/op  0 allocs/op

# BenchmarkEvenChannel-8    1560124  1249 ns/op  158 B/op  0 allocs/op

代码( main_test.go):

package main

import (

    "flag"

    "fmt"

    "os"

    "sync"

    "testing"

)

func BenchmarkEvenWaitgroup(b *testing.B) {

    evenWaitgroup(b.N)

}

func BenchmarkEvenChannel(b *testing.B) {

    evenChannel(b.N)

}

func evenWaitgroup(n int) {

    if n%2 == 1 { // make it even:

        n++

    }

    for i := 0; i < n; i++ {

        wg.Add(1)

        go func(n int) {

            select {

            case ch <- n: // tx if channel is empty

            case i := <-ch: // rx if channel is not empty

                // fmt.Println(n, i)

                _ = i

            }

            wg.Done()

        }(i)

    }

    wg.Wait()

}

func evenChannel(n int) {

    if n%2 == 1 { // make it even:

        n++

    }

    for i := 0; i < n; i++ {

        go func(n int) {

            select {

            case ch <- n: // tx if channel is empty

            case i := <-ch: // rx if channel is not empty

                // fmt.Println(n, i)

                _ = i

            }

            done <- struct{}{}

        }(i)

    }

    for i := 0; i < n; i++ {

        <-done

    }

}

func TestMain(m *testing.M) {

    var n int // We use TestMain to set up the done channel.

    flag.IntVar(&n, "n", 1_000_000, "chan cap")

    flag.Parse()

    done = make(chan struct{}, n)

    fmt.Println("n=", n)

    os.Exit(m.Run())

}

var (

    done chan struct{}

    ch   = make(chan int)

    wg   sync.WaitGroup

)