考虑这个函数:

func doAllWork() error {

    var wg sync.WaitGroup

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

        wg.add(1)
        go func() {

            defer wg.Done()
            for j := 0; j < 10; j++ {
                result, err := work(j)
                if err != nil {
                    // can't use `return err` here
                    // what sould I put instead ? 
                    os.Exit(0)
                }
            }
        }()
    }
    wg.Wait()

    return nil
}

在每个 goroutine 中，函数 work() 被调用了 10 次。如果对 work() 的调用在任何正在运行的 goroutine 中返回错误，我希望所有 goroutine 立即停止，并退出程序。 可以在这里使用 os.Exit() 吗？我该如何处理？

编辑:这个问题不同于how to stop a goroutine在这里，如果一个 goroutine 发生错误，我需要关闭所有 goroutine

最佳答案

您可以使用 context为这样的事情创建的包(“带有截止日期，取消信号......”)。

您创建一个能够发布取消信号的上下文 context.WithCancel() (父上下文可能是由 context.Background() 返回的上下文)。这将向您返回一个 cancel() 函数，该函数可用于取消(或更准确地说是发出取消意图的信号)给 worker goroutines。
在 worker goroutines 中，您必须通过检查 Context.Done() 返回的 channel 是否关闭来检查是否已启动此类意图，最简单的方法是尝试从中接收(立即进行)如果关闭)。要进行非阻塞检查(如果它未关闭，您可以继续)，使用带有 default 分支的 select 语句。

我将使用以下 work() 实现，它模拟 10% 的失败几率，并模拟 1 秒的工作:

func work(i int) (int, error) {
    if rand.Intn(100) < 10 { // 10% of failure
        return 0, errors.New("random error")
    }
    time.Sleep(time.Second)
    return 100 + i, nil
}

doAllWork() 可能如下所示:

func doAllWork() error {
    var wg sync.WaitGroup

    ctx, cancel := context.WithCancel(context.Background())
    defer cancel() // Make sure it's called to release resources even if no errors

    for i := 0; i < 2; i++ {
        wg.Add(1)
        go func(i int) {
            defer wg.Done()

            for j := 0; j < 10; j++ {
                // Check if any error occurred in any other gorouties:
                select {
                case <-ctx.Done():
                    return // Error somewhere, terminate
                default: // Default is must to avoid blocking
                }
                result, err := work(j)
                if err != nil {
                    fmt.Printf("Worker #%d during %d, error: %v\n", i, j, err)
                    cancel()
                    return
                }
                fmt.Printf("Worker #%d finished %d, result: %d.\n", i, j, result)
            }
        }(i)
    }
    wg.Wait()

    return ctx.Err()
}

这是如何测试它的:

func main() {
    rand.Seed(time.Now().UnixNano() + 1) // +1 'cause Playground's time is fixed
    fmt.Printf("doAllWork: %v\n", doAllWork())
}

输出(在 Go Playground 上尝试):

Worker #0 finished 0, result: 100.
Worker #1 finished 0, result: 100.
Worker #1 finished 1, result: 101.
Worker #0 finished 1, result: 101.
Worker #0 finished 2, result: 102.
Worker #1 finished 2, result: 102.
Worker #1 finished 3, result: 103.
Worker #1 during 4, error: random error
Worker #0 finished 3, result: 103.
doAllWork: context canceled

如果没有错误，例如使用以下 work() 函数时:

func work(i int) (int, error) {
    time.Sleep(time.Second)
    return 100 + i, nil
}

输出就像(在 Go Playground 上尝试):

Worker #0 finished 0, result: 100.
Worker #1 finished 0, result: 100.
Worker #1 finished 1, result: 101.
Worker #0 finished 1, result: 101.
Worker #0 finished 2, result: 102.
Worker #1 finished 2, result: 102.
Worker #1 finished 3, result: 103.
Worker #0 finished 3, result: 103.
Worker #0 finished 4, result: 104.
Worker #1 finished 4, result: 104.
Worker #1 finished 5, result: 105.
Worker #0 finished 5, result: 105.
Worker #0 finished 6, result: 106.
Worker #1 finished 6, result: 106.
Worker #1 finished 7, result: 107.
Worker #0 finished 7, result: 107.
Worker #0 finished 8, result: 108.
Worker #1 finished 8, result: 108.
Worker #1 finished 9, result: 109.
Worker #0 finished 9, result: 109.
doAllWork: <nil>

注意事项:

基本上我们只是使用上下文的 Done() channel ，所以看起来我们可以很容易地(如果不是更容易的话)使用 done channel 而不是Context，关闭 channel 以执行上述解决方案中 cancel() 的操作。

这不是真的。 这只能在只有一个 goroutine 可以关闭 channel 的情况下使用，但在我们的例子中，任何一个 worker 都可以这样做。 并试图关闭一个已经关闭的 channel 会出现 panic (详情请参见此处:How does a non initialized channel behave? ).因此，您必须确保围绕 close(done) 进行某种同步/排除，这将使其可读性降低甚至更加复杂。实际上，这正是 cancel() 函数在幕后所做的，隐藏/抽象远离您的眼睛，因此 cancel() 可能会被多次调用以使您的代码/使用起来更简单。

如何获取并返回worker的错误信息？

为此你可以使用错误 channel :

errs := make(chan error, 2) // Buffer for 2 errors

当遇到错误时，在 worker 内部，将错误发送到 channel 而不是打印出来:

result, err := work(j)
if err != nil {
    errs <- fmt.Errorf("Worker #%d during %d, error: %v\n", i, j, err)
    cancel()
    return
}

在循环之后，如果出现错误，则返回错误(否则返回 nil):

// Return (first) error, if any:
if ctx.Err() != nil {
    return <-errs
}
return nil

这次输出(在 Go Playground 上试试):

Worker #0 finished 0, result: 100.
Worker #1 finished 0, result: 100.
Worker #1 finished 1, result: 101.
Worker #0 finished 1, result: 101.
Worker #0 finished 2, result: 102.
Worker #1 finished 2, result: 102.
Worker #1 finished 3, result: 103.
Worker #0 finished 3, result: 103.
doAllWork: Worker #1 during 4, error: random error

请注意，我使用了一个缓冲 channel ，其缓冲区大小等于工作人员的数量，这确保在其上发送始终是非阻塞的。这也使您有可能接收和处理所有错误，而不仅仅是一个(例如第一个)。另一种选择可能是使用缓冲 channel 仅容纳 1，并对其进行非阻塞发送，如下所示:

errs := make(chan error, 1) // Buffered only for the first error

// ...and inside the worker:

result, err := work(j)
if err != nil {
    // Non-blocking send:
    select {
    case errs <- fmt.Errorf("Worker #%d during %d, error: %v\n", i, j, err):
    default:
    }
    cancel()
    return
}

关于go - 一个go中出错关闭多个goroutine，我们在Stack Overflow上找到一个类似的问题： https://stackoverflow.com/questions/56754953/