所有 goroutine 都在我的异步代码中休眠

异步代码中，当所有协程都休眠时，程序将陷入死锁。为了解决这个问题，需要使用同步机制，例如 `sync.WaitGroup` 或 `errgroup.WithContext`。这些机制允许在协程完成执行后发出信号，以便主协程知道所有协程都已完成任务，从而避免死锁的发生。

我读过这个、这个和这个，但没有一个能解决我的问题..

我正在尝试异步读取 2 个文件，因此我编写了以下内容：

//readlines.go
package main

import (
    "bufio"
    "os"
)

// readlines reads a whole file into memory
// and returns a slice of its lines.
func readlines(path string) ([]string, error) {
    file, err := os.open(path)
    if err != nil {
        return nil, err
    }
    defer file.close()

    var lines []string
    scanner := bufio.newscanner(file)
    for scanner.scan() {
        lines = append(lines, scanner.text())
    }
    return lines, scanner.err()
}

并将其称为：

package main

import (
    "fmt"
    "os"

    "github.com/gocarina/gocsv"
)

func (s *stocks) read() {
    fmt.println("reading")
    stockfile, err := os.openfile("current_invenory.csv", os.o_rdwr|os.o_create, os.modeperm)
    if err != nil {
        panic(err)
    }
    defer stockfile.close()
    stocks := []systemstock{}
    if err := gocsv.unmarshalfile(stockfile, &stocks); err != nil { // load stocks from file
        panic(err)
    }

    *s = stocks
}

package main

import (
    "fmt"
    "os"

    "github.com/gocarina/gocsv"
)

func (t *transactions) read() {
    fmt.println("reading")
    trxfile, err := os.openfile("current_transactions.csv", os.o_rdwr|os.o_create, os.modeperm)
    if err != nil {
        panic(err)
    }
    defer trxfile.close()
    trx := []systemtransactions{}
    if err := gocsv.unmarshalfile(trxfile, &trx); err != nil { // load stocks from file
        panic(err)
    }

    *t = trx
}

上面的工作非常好：

    stock := stocks{} 
    trx := transactions{}

    stock.read()
    trx.read()
    for _, s := range stock {
            fmt.println("hello", s.code)
    }

但是当我尝试将它们读取为：时给出错误 fatal error: all goroutine are sleep - deadlock!:

    cs, ct := readdata()

    for _, s := range cs {
        fmt.println("hello", s.code)
    }

    for _, t := range ct {
        fmt.println("hello trx of ", t.code)
    }

使用

import "sync"

//func readData(cs chan stocks, ct chan transactions) (stocks, transactions) {
func readData() (stocks, transactions) {
    var wg sync.WaitGroup
    defer wg.Done()

    stock := stocks{}
    trx := transactions{}

    wg.Add(1)
    go stock.Read()
    wg.Add(1)
    go trx.Read()

    wg.Wait()

    return stock, trx
}

所以这个错误与我在最后一个块中犯的错误（或不理解）有关〜

解决方案

要同时运行 stocks 和 transactions 的 read 方法，这些方法需要有一种在执行完成时发出信号的方式。这可以通过多种方式完成，但这里有两种需要对代码进行最少修改的方式。

解决方案1

使用 sync.waitgroup 包。使用此包， read 方法在完成执行后应执行 wg.done() 语句。它应该看起来像这样：

func (s *stocks) read(wg *sync.waitgroup) {
    defer wg.done()
    fmt.println("reading")
    stockfile, err := os.openfile("current_invenory.csv", os.o_rdwr|os.o_create, os.modeperm)
    if err != nil {
        panic(err)
    }
    defer stockfile.close()
    stocks := []systemstock{}
    if err := gocsv.unmarshalfile(stockfile, &stocks); err != nil { // load stocks from file
        panic(err)
    }

    *s = stocks
}

func (t *transactions) read(wg *sync.waitgroup) {
    defer wg.done()
    fmt.println("reading")
    trxfile, err := os.openfile("current_transactions.csv", os.o_rdwr|os.o_create, os.modeperm)
    if err != nil {
        panic(err)
    }
    defer trxfile.close()
    trx := []systemtransactions{}
    if err := gocsv.unmarshalfile(trxfile, &trx); err != nil { // load stocks from file
        panic(err)
    }

    *t = trx
}

func readdata() (stocks, transactions) {
    var wg sync.waitgroup
    wg.add(2)

    stock := stocks{}
    trx := transactions{}

    go stock.read(&wg)
    go trx.read(&wg)

    wg.wait()

    return stock, trx
}

解决方案2

此方法使用 golang.org/x/sync/errgroup 包。在这种情况下，您不需要自己处理同步和信令，但使用 errgroup.go 方法添加的函数需要具有严格的 func() error 签名。您的代码应如下所示：

func (s *stocks) read() error {
        fmt.println("reading")
        stockfile, err := os.openfile("current_invenory.csv", os.o_rdwr|os.o_create, os.modeperm)
        if err != nil {
            return err
        }
        defer stockfile.close()
        stocks := []systemstock{}
        if err := gocsv.unmarshalfile(stockfile, &stocks); err != nil { // load stocks from file
            return err
        }
    
        *s = stocks
        return nil
    }
    
    func (t *transactions) read() error {
        fmt.println("reading")
        trxfile, err := os.openfile("current_transactions.csv", os.o_rdwr|os.o_create, os.modeperm)
        if err != nil {
            return err
        }
        defer trxfile.close()
        trx := []systemtransactions{}
        if err := gocsv.unmarshalfile(trxfile, &trx); err != nil { // load stocks from file
            return err
        }
    
        *t = trx
        return nil
    }
    
    func readdata() (stocks, transactions) {
        g, _ := errgroup.withcontext(context.background())
    
        stock := stocks{}
        trx := transactions{}
    
        g.go(stock.read)
        g.go(trx.read)
    
        if err:= g.wait(); err != nil {
           panic(err)
        }
    
        return stock, trx
    }

解决方案3

当您开始从每个 csv 读取数据时，您（正确地）向等待组添加 1，从而将等待组的内部计数器设置为 2，但是 wg.wait() 会等到该计数器降至零并且您不会这样做没有任何调用 wg.done() 来做到这一点。我建议将 go stock.read() 更改为：

go func() {
    defer wg done()
    stock.read()
}()

所以，完整的工作代码是：

func readData() (stocks, transactions) {
    var wg sync.WaitGroup

    stock := stocks{}
    trx := transactions{}

    wg.Add(1)
    go func() {
        defer wg.Done()
        stock.Read()
    }()
    wg.Add(1)
    go func() {
        defer wg.Done()
        trx.Read()
    }()

    wg.Wait()

    return stock, trx
}

本篇关于《所有 goroutine 都在我的异步代码中休眠》的介绍就到此结束啦，但是学无止境，想要了解学习更多关于Golang的相关知识，请关注golang学习网公众号！