如何解决生产者和消费者代码中的死锁

Question

当我运行下面的程序时，出现错误davecheney      tweets about golangbeertocode      does not tweet about golangironzeb         tweets about golangbeertocode      tweets about golangvampirewalk666  tweets about golangfatal error: all goroutines are asleep - deadlock!goroutine 1 [semacquire]:sync.runtime_Semacquire(0xc000010260?)        /usr/local/go/src/runtime/sema.go:56 +0x25sync.(*WaitGroup).Wait(0x100c000058058?)        /usr/local/go/src/sync/waitgroup.go:136 +0x52main.main()        /home/joe/go/src/github.com/go-concurrency-exercises/1-producer-consumer/main.go:53 +0x14f死锁从何而来，如何改进程序以避免死锁？package mainimport (    "fmt"    "sync"    "time")func producer(stream Stream, tweetChan chan *Tweet) {    for {        tweet, err := stream.Next()        if err == ErrEOF {            close(tweetChan)            return        }        tweetChan <- tweet        //tweets = append(tweets, tweet)    }}func consumer(tweetChan chan *Tweet) {    for t := range tweetChan {        if t.IsTalkingAboutGo() {            fmt.Println(t.Username, "\ttweets about golang")        } else {            fmt.Println(t.Username, "\tdoes not tweet about golang")        }    }}func main() {    start := time.Now()    stream := GetMockStream()    var wg sync.WaitGroup    tweetChan := make(chan *Tweet)    // Producer    //tweets := producer(stream)    wg.Add(2)    go producer(stream, tweetChan)    // Consumer    //consumer(tweets)    go consumer(tweetChan)    wg.Wait()    fmt.Printf("Process took %s\n", time.Since(start))}如果需要看mockstream.go，参考 https://github.com/loong/go-concurrency-exercises/tree/master/1-producer-consumer我的程序是原程序修改main.go的并发版本

Answer 1

对 wg.Wait() 的调用一直在等待，直到组的计数器为零，但是没有正在运行的 goroutines 来递减计数器。

通过在从 goroutine 函数返回之前调用 wg.Done() 来修复：

func producer(wg *sync.WaitGroup, stream Stream, tweetChan chan *Tweet) {

    defer wg.Done()

    for {

        tweet, err := stream.Next()

        if err == ErrEOF {

            close(tweetChan)

            return

        }

        tweetChan <- tweet

    }

}

func consumer(wg *sync.WaitGroup, tweetChan chan *Tweet) {

    defer wg.Done()

    for t := range tweetChan {

        if t.IsTalkingAboutGo() {

            fmt.Println(t.Username, "\ttweets about golang")

        } else {

            fmt.Println(t.Username, "\tdoes not tweet about golang")

        }

    }

}

func main() {

    start := time.Now()

    stream := GetMockStream()

    var wg sync.WaitGroup

    tweetChan := make(chan *Tweet)

    wg.Add(2)

    go producer(&wg, stream, tweetChan)

    go consumer(&wg, tweetChan)

    wg.Wait()

    fmt.Printf("Process took %s\n", time.Since(start))

}