缓冲区为空后关闭“worker”go 例程

我想要我的 goroutine 工作人员（ProcessToDo()在下面的代码中）等待所有“排队”工作处理完毕后再关闭。

工作例程有一个“待办事项”通道（缓冲），通过该通道将工作发送给它。它有一个“完成”通道来告诉它开始关闭。文档称，如果满足多个选择，通道上的选择将选择一个“伪随机值”...这意味着在所有缓冲工作完成之前将触发关闭（返回）。

在下面的代码示例中，我希望打印所有 20 条消息...

package main

import (
    "time"
    "fmt"
)


func ProcessToDo(done chan struct{}, todo chan string) {
    for {
        select {
        case work, ok := <-todo:
            if !ok {
                fmt.Printf("Shutting down ProcessToDo - todo channel closed!\n")
                return
            }
            fmt.Printf("todo: %q\n", work)
            time.Sleep(100 * time.Millisecond)
        case _, ok := <-done:
            if ok {
                fmt.Printf("Shutting down ProcessToDo - done message received!\n")
            } else {
                fmt.Printf("Shutting down ProcessToDo - done channel closed!\n")
            }
            close(todo)
            return
        }
    }
}

func main() {

    done := make(chan struct{})
    todo := make(chan string, 100)

    go ProcessToDo(done, todo)

    for i := 0; i < 20; i++ {
        todo <- fmt.Sprintf("Message %02d", i)
    }

    fmt.Println("*** all messages queued ***")
    time.Sleep(1 * time.Second)
    close(done)
    time.Sleep(4 * time.Second)
}

done在您的情况下，通道是完全不必要的，因为您可以通过关闭来发出关闭信号todo频道本身。

并使用for range在通道上，它将迭代直到通道关闭并且其缓冲区为空。

你应该有一个done通道，但只是为了让 goroutine 本身可以发出它已完成工作的信号，以便主 goroutine 可以继续或退出。

这个变体与你的相同，更简单并且不需要time.Sleep()调用等待其他 goroutine（无论如何这都太错误且不确定）。尝试一下去游乐场:

func ProcessToDo(done chan struct{}, todo chan string) {
    for work := range todo {
        fmt.Printf("todo: %q\n", work)
        time.Sleep(100 * time.Millisecond)
    }
    fmt.Printf("Shutting down ProcessToDo - todo channel closed!\n")
    done <- struct{}{} // Signal that we processed all jobs
}

func main() {
    done := make(chan struct{})
    todo := make(chan string, 100)

    go ProcessToDo(done, todo)

    for i := 0; i < 20; i++ {
        todo <- fmt.Sprintf("Message %02d", i)
    }

    fmt.Println("*** all messages queued ***")
    close(todo)
    <-done // Wait until the other goroutine finishes all jobs
}

另请注意，工作协程应使用以下方式发出完成信号：defer因此，如果worker以某种意外的方式返回，或者出现恐慌，主goroutine不会陷入等待worker的状态。所以它应该这样开始：

defer func() {
    done <- struct{}{} // Signal that we processed all jobs
}()

您还可以使用sync.WaitGroup将主 goroutine 同步到工作线程（等待它）。事实上，如果您计划使用多个工作协程，那么这比从done渠道。而且用以下命令来表示完成也更简单WaitGroup因为它有一个Done()方法（这是一个函数调用），所以你不需要匿名函数：

defer wg.Done()

See 吉姆博的回答完整的例子WaitGroup.

使用for range如果您想使用多个工作协程，这也是惯用的：通道是同步的，因此您不需要任何额外的代码来同步对todo频道或从中收到的作业。如果你关闭todo频道中的main()，这将正确地向所有工作协程发出信号。当然，所有排队的作业都会被接收并处理一次。

现在采用使用的变体WaitGroup让主 Goroutine 等待工人（JimB 的回答）：如果你想要超过 1 个工人 Goroutine 怎么办？同时处理您的工作（并且很可能是并行的）？

您需要在代码中添加/更改的唯一内容是：要真正启动其中多个：

for i := 0; i < 10; i++ {
    wg.Add(1)
    go ProcessToDo(todo)
}

无需更改任何其他内容，您现在就有了一个正确的并发应用程序，它使用 10 个并发 goroutine 接收并处理您的作业。而且我们没有使用任何“丑陋”的东西time.Sleep()（我们使用了一个，但只是为了模拟缓慢的处理，而不是等待其他 goroutine），并且您不需要任何额外的同步。