如何将渠道和等待组的工作结合起来？

亲爱的编程学习爱好者，如果你点开了这篇文章，说明你对《如何将渠道和等待组的工作结合起来？》很感兴趣。本篇文章就来给大家详细解析一下，主要介绍一下，希望所有认真读完的童鞋们，都有实质性的提高。

有一个代码：

func main() {
    rand.Seed(time.Now().UnixNano())
    res, err := MockFunc()
    fmt.Printf("res - %v, err - %v\n", res, err)
}

func MockFunc() ([]int, error) {
    args := []int{1, 2, 3, 4, 5}
    result := make([]int, 0)
    errCh := make(chan error)

    wg := &sync.WaitGroup{}
    wg.Add(len(args))
    for _, a := range args {
        go func(aLit int) {
            value, err := RandomError(aLit)
            if err != nil {
                errCh <- err
                return
            }
            result = append(result, value)
        }(a)
    }
    errValue := <-errCh // if no errors, i have panic - fatal error: all goroutines are asleep - deadlock!
    if errValue != nil {
        fmt.Println("returning because err")
        return nil, errValue // if RandomError() returns error, returned here as expected
    }
    wg.Wait()
    return result, nil
}

func RandomError(arg int) (int, error) {
    time.Sleep(time.Millisecond * 100 * time.Duration(arg))

    errChance := rand.Intn(100)
    if errChance > 40 {
        fmt.Printf("error on arg - %d\n", arg)
        return 0, errors.New("mock err")
    }

    return errChance, nil
}

我需要，如果 randomerror() 函数返回错误，则 mockfunc() 函数在等待所有 waitgroups 并返回错误之前完成。但如果没有错误，那么我会得到一个 deadlock，如果有，那么一切都会按预期进行。

我明白这是因为我没有关闭频道。但是如果我在 wg.wait() 之后关闭它，那么它的意义就会丢失，因为如果第一次调用该函数返回错误，那么我将等待所有其他调用的结果。

我需要它，以便如果对 randomeerr() 的其中一个调用返回错误，我会从 mockfunc() 返回此错误到 main()，而无需等待所有调用结束。

正确答案

使用 errgroup.group 可能更容易。

我还必须创建另一个 goroutine 和通道来将结果读入数组。如果您在许多 goroutine 中使用 result =append(result, value)，您有时会收到错误。

使用 errgroup.group 的好处之一是在函数返回之前必须清理并停止所有 goroutine。当这些类型的函数是较大系统的一部分时，这有助于防止内存泄漏。

package main

import (
    "context"
    "errors"
    "fmt"
    "math/rand"
    "time"

    "golang.org/x/sync/errgroup"
)

func main() {
    rand.seed(time.now().unixnano())
    res, err := mockfunc()
    fmt.printf("res - %v, err - %v\n", res, err)
}

func mockfunc() ([]int, error) {
    args := []int{1, 2, 3, 4, 5}
    resultchan := make(chan int)
    result := make([]int, 0)
    eg, ctx := errgroup.withcontext(context.background())

    // read all results into an array
    eg.go(func() error {
        for {
            select {
            case <-ctx.done():
                // always check to see if the context has cancelled,
                // if there is an error errgroup will cancel the
                // context and all goroutines will need to exit
                // before `eg.wait` returns.
                return context.canceled
            case val := <-resultchan:
                result = append(result, val)
                if len(result) == len(args) {
                    return nil
                }
            }
        }
    })

    for _, a := range args {
        alit := a // copy the value so that we don't re-use the memory address
        eg.go(func() error {
            value, err := randomerror(alit)
            if err != nil {
                return err
            }
            select {
            case resultchan <- value:
            case <-ctx.done():
                return context.canceled
            default:
            }
            return nil
        })
    }
    return result, eg.wait()
}

func randomerror(arg int) (int, error) {
    time.sleep(time.millisecond * 100 * time.duration(arg))

    errchance := rand.intn(100)
    if errchance > 40 {
        fmt.printf("error on arg - %d\n", arg)
        return 0, errors.new("mock err")
    }

    return errchance, nil
}

如果将所有结果缓冲到一个通道中也会更简单：

func mockfunc() ([]int, error) {
    args := []int{1, 2, 3, 4, 5}
    resultchan := make(chan int, len(args))
    result := make([]int, 0)
    eg, ctx := errgroup.withcontext(context.background())

    for _, a := range args {
        alit := a // copy the value so that we don't re-use the memory address
        eg.go(func() error {
            value, err := randomerror(alit)
            if err != nil {
                return err
            }
            select {
            case resultchan <- value:
            case <-ctx.done():
                return context.canceled
            default:
            }
            return nil
        })
    }
    if err := eg.wait(); err != nil {
        return nil, err
    }
    close(resultchan)
    for val := range resultchan {
        result = append(result, val)
    }
    return result, nil
}

这是我能想到的最简单的实现：

func mockfunc() (result []int, err error) {
    args := []int{1, 2, 3, 4, 5}
    resultchan := make(chan int)
    errchan := make(chan error)
    for _, a := range args {
        go func(alit int) {
            value, err := randomerror(alit)
            if err != nil {
                errchan <- err
            }
            resultchan <- value
        }(a)
    }
    for i := 0; i < len(args); i++ {
        select {
        case err := <-errchan:
            return nil, err
        case val := <-resultchan:
            result = append(result, val)
        }
    }
    return result, nil
}

您遇到了三个死锁：

1. errvalue := <-errch // 如果没有错误，我就会恐慌 - 致命错误：所有 goroutine 都在睡觉 - 死锁！
2. wg.wait() # 第 44 行
3. errch <- err # 第 29 行

由于 return 语句或第一个死锁，第二个从未发生。 为了修复死锁1，我们必须以非阻塞模式读取通道。 为了修复死锁 2，我们必须在每个 goroutine 中使用 wg.done。 为了修复死锁 3，我们必须使用缓冲通道，或者以某种方式我们必须消耗该通道。为简单起见，这里我选择缓冲通道，但我们可以在 for 循环内读取 errch 并在看到错误时返回错误。

package main

import (
    "errors"
    "fmt"
    "math/rand"
    "sync"
    "time"
)

func main() {
    rand.Seed(time.Now().UnixNano())
    res, err := MockFunc()
    fmt.Printf("res - %v, err - %v\n", res, err)
}

func MockFunc() ([]int, error) {
    args := []int{1, 2, 3, 4, 5}
    result := make([]int, 0)
    errCh := make(chan error, len(args))

    wg := &sync.WaitGroup{}
    wg.Add(len(args))
    for _, a := range args {
        go func(aLit int) {
            defer wg.Done()
            value, err := RandomError(aLit)
            if err != nil {
                errCh <- err
                return
            }
            result = append(result, value)
        }(a)
    }
    var errValue error
    select {
    case errValue = <-errCh: // if no errors, i have panic - fatal error: all goroutines are asleep - deadlock!
    default:
    }
    if errValue != nil {
        fmt.Println("returning because err")
        return nil, errValue // if RandomError() returns error, returned here as expected
    }
    wg.Wait()
    return result, nil
}

func RandomError(arg int) (int, error) {
    time.Sleep(time.Millisecond * 100 * time.Duration(arg))

    errChance := rand.Intn(100)
    if errChance > 40 {
        fmt.Printf("error on arg - %d\n", arg)
        return 0, errors.New("mock err")
    }

    return errChance, nil
}

终于介绍完啦！小伙伴们，这篇关于《如何将渠道和等待组的工作结合起来？》的介绍应该让你收获多多了吧！欢迎大家收藏或分享给更多需要学习的朋友吧~golang学习网公众号也会发布Golang相关知识，快来关注吧！