如何使用 WaitGroup 处理工作程序池中的错误？

Question

同时使用 sync.WaitGroup 和 select 时遇到问题。如果您查看以下 http 请求池，您会注意到如果发生错误，它将永远不会被报告，因为wg.Done() 将阻塞并且不再从通道读取。

package pool

import (
    "fmt"
    "log"
    "net/http"
    "sync"
)

var (
    MaxPoolQueue  = 100
    MaxPoolWorker = 10
)

type Pool struct {
    wg *sync.WaitGroup

    queue  chan *http.Request
    errors chan error
}

func NewPool() *Pool {
    return &Pool{
        wg: &sync.WaitGroup{},

        queue:  make(chan *http.Request, MaxPoolQueue),
        errors: make(chan error),
    }
}

func (p *Pool) Add(r *http.Request) {
    p.wg.Add(1)

    p.queue <- r
}

func (p *Pool) Run() error {
    for i := 0; i < MaxPoolWorker; i++ {
        go p.doWork()
    }

    select {
    case err := <-p.errors:
        return err
    default:
        p.wg.Wait()
    }

    return nil
}

func (p *Pool) doWork() {
    for r := range p.queue {
        fmt.Printf("Request to %s\n", r.Host)

        p.wg.Done()

        _, err := http.DefaultClient.Do(r)

        if err != nil {
            log.Fatal(err)

            p.errors <- err
        } else {
            fmt.Printf("no error\n")
        }
    }
}

来源可以找到here

我怎样才能在使用 WaitGroup 的同时从 go 例程中得到错误？

Answer 1

我在写这个问题时自己得到了答案，因为我认为这是一个有趣的案例，所以我想与你分享。

同时使用sync.WaitGroup 和chan 的诀窍是我们包装：

select {
    case err := <-p.errors:
        return err
    default:
        p.wg.Done()
}

一起在for 循环中：

for {
    select {
        case err := <-p.errors:
            return err
        default:
            p.wg.Done()
    }
}

在这种情况下，select 将始终检查错误并等待如果没有任何反应:)

Answer 2

它看起来有点像Tomb library (Tomb V2 GoDoc) 启用的快速失败机制：

tomb 包处理干净的 goroutine 跟踪和终止。

如果任何被跟踪的 goroutine 返回非 nil 错误，或者系统中的任何 goroutine（是否跟踪）调用了 Kill 或 Killf 方法，则设置墓穴 Err，@987654327 @ 设置为 false，Dying 通道关闭以标记所有被跟踪的 goroutine 都应该自愿终止。

一旦所有被跟踪的 goroutine 终止，Dead 通道将关闭，Wait 解除阻塞并通过结果或显式 Kill 或 Killf 方法调用返回第一个非零错误呈现给 tomb，如果没有错误，则为 nil。

你可以看一个例子in this playground:

（摘录）

// start runs all the given functions concurrently
// until either they all complete or one returns an
// error, in which case it returns that error.
//
// The functions are passed a channel which will be closed
// when the function should stop.
func start(funcs []func(stop <-chan struct{}) error) error {
    var tomb tomb.Tomb
    var wg sync.WaitGroup
    allDone := make(chan struct{})
    // Start all the functions.
    for _, f := range funcs {
        f := f
        wg.Add(1)
        go func() {
            defer wg.Done()
            if err := f(tomb.Dying()); err != nil {
                tomb.Kill(err)
            }
        }()
    }
    // Start a goroutine to wait for them all to finish.
    go func() {
        wg.Wait()
        close(allDone)
    }()

    // Wait for them all to finish, or one to fail
    select {
    case <-allDone:
    case <-tomb.Dying():
    }
    tomb.Done()
    return tomb.Err()
}

Answer 3

一个更简单的实现如下所示。 （签入play.golang：https://play.golang.org/p/TYxxsDRt5Wu）

package main

import "fmt"
import "sync"
import "time"

type Error struct {
    message string
}

func (e Error) Error() string {
    return e.message
}

func main() {
    var wg sync.WaitGroup
    waitGroupLength := 8
    errChannel := make(chan error, 1)

    // Setup waitgroup to match the number of go routines we'll launch off
    wg.Add(waitGroupLength)
    finished := make(chan bool, 1) // this along with wg.Wait() are why the error handling works and doesn't deadlock

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

        go func(i int) {
            fmt.Printf("Go routine %d executed\n", i+1)
            time.Sleep(time.Duration(waitGroupLength - i))
            time.Sleep(0) // only here so the time import is needed
            
            if i%4 == 1 {
                errChannel <- Error{fmt.Sprintf("Errored on routine %d", i+1)}
            }
            
            // Mark the wait group as Done so it does not hang
            wg.Done()
        }(i)
    }

    go func() {
        wg.Wait()
        close(finished)
    }()

    L:
        for {
            select {
            case <-finished:
                break L // this will break from loop
            case err := <-errChannel:
                if err != nil {
                    fmt.Println("error ", err)
                    // handle your error
                }
            }
        }

    fmt.Println("Executed all go routines")
}