使用 channnels 的 goroutines 同步问题

Question

我正在使用以下代码来同步goroutines。最近在调查一个错误时，我发现下面的代码并不总是有效。大约五分之一的失败。频道在我的频道之前获取消息。我能够在本地（而不是在go-playground）和k8s环境中始终如一地重现此问题。作为一种解决方法，我现在使用同步。quitoutsync.Map有没有办法修复下面的代码？package mainimport (    "fmt"    "io/ioutil"    "log"    "os"    "path"    "sync"    "sync/atomic"    "time")func main() {    //test setup    filePaths := []string{        path.Join(os.TempDir(), fmt.Sprint("f1-", time.Now().Nanosecond())),        path.Join(os.TempDir(), fmt.Sprint("f2-", time.Now().Nanosecond())),        path.Join(os.TempDir(), fmt.Sprint("f3-", time.Now().Nanosecond())),        path.Join(os.TempDir(), fmt.Sprint("f4-", time.Now().Nanosecond())),        path.Join(os.TempDir(), fmt.Sprint("f5-", time.Now().Nanosecond())),        path.Join(os.TempDir(), fmt.Sprint("f6-", time.Now().Nanosecond())),    }    for _, filePath := range filePaths {        f, err := os.OpenFile(filePath, os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0600)        if err != nil {            log.Fatal(err)        }        _, err = f.WriteString("There are many variations of passages of Lorem Ipsum available, but the majority have suffered alteration in some form, by injected humour, or randomised words which don't look even slightly believable. If you are going to use a passage of Lorem Ipsum, you need to be sure there isn't anything embarrassing hidden in the middle of text. All the Lorem Ipsum generators on the Internet tend to repeat predefined chunks as necessary, making this the first true generator on the Internet. It uses a dictionary of over 200 Latin words, combined with a handful of model sentence structures, to generate Lorem Ipsum which looks reasonable. The generated Lorem Ipsum is therefore always free from repetition, injected humour, or non-characteristic words etc.")        if err != nil {            log.Fatal(err)        }        err = f.Close()        if err != nil {            log.Fatal(err)        }    }

Answer 1

收到退出时，输出通道可以包含值。通过制作无缓冲通道进行修复：

out := make(chan []byte)

这可确保在退出之前收到从工作线程发送的值：

• 在无缓冲信道上发送/接收发生在调用wg.Done()

• 所有电话在返回之前发生wg.Done()wg.Wait()

• wg.Wait()在将值发送到 之前返回quit

因此，在将值发送到 之前，将从 接收值。outquit

另一种方法是关闭通道，向结果收集器发出工作线程已完成的信号：out

func getContents(fileNames []string) ([][]byte, error) {

    wg := sync.WaitGroup{}

    var responseBytes [][]byte

    out := make(chan []byte)

    var opsFileRead uint64

    var opsChannelGot uint64

    for _, fileName := range fileNames {

        wg.Add(1)

        go func(fName string, out chan []byte, wg *sync.WaitGroup) {

            defer wg.Done()

            data, err := ioutil.ReadFile(fName)

            if err != nil {

                log.Fatal(err)

            }

            out <- data

            atomic.AddUint64(&opsFileRead, 1)

        }(fileName, out, &wg)

    }

    // Close out after workers are done.

    go func() {

        wg.Wait()

        close(out)

    }()

    // Loop over outputs until done.

    for bts := range out {

        if len(bts) > 0 {

            atomic.AddUint64(&opsChannelGot, 1)

            responseBytes = append(responseBytes, bts)

        }

    }

    fmt.Printf("I quit, i read %d, i got %d\n", opsFileRead, opsChannelGot)

    return responseBytes, nil

}