multithreading - 所有go routines are sleeve deadlock

Question

我正在学习如何进行并发，我已经将其编写为自己的应用程序，以便在它运行后可以将其移植到不同的项目中。

我要添加它的项目基本上会将 RowInfo 发送到全局 QueueChannel，然后我的工作人员应该接手这项工作并进行处理。如果我将具有相同 ID 的两行排入队列，并且其中一行当前正在由工作人员处理，我将从队列中删除重复的行(如您所见，我在调度程序中执行“继续”的位置)。

此排队/工作人员代码将在 ListenAndServe 阻塞的 Web 服务器上运行，因此我希望它始终保持运行状态，并且工作人员始终积极寻找工作。我不想关闭 channel (除非我 ctrl+C'd 应用程序或其他东西)。我怀疑我遇到的错误与未关闭 channel 有关，因为许多其他提到此错误的线程似乎都表明了这一点，但我不确定它与我所拥有的代码有何关系。

终端错误输出:

[~/go/src/github.com/zzz/asynch]> go run main.go
fatal error: all goroutines are asleep - deadlock!

goroutine 1 [chan send]:
main.main()
    /home/zzz/go/src/github.com/zzz/asynch/main.go:29 +0x14b

goroutine 5 [select]:
main.diszzzcher(0xc82001a120, 0xc82001a180, 0xc82001a1e0)
    /home/zzz/go/src/github.com/zzz/asynch/main.go:42 +0x21a
created by main.main
    /home/zzz/go/src/github.com/zzz/asynch/main.go:19 +0xb1

goroutine 6 [chan receive]:
main.worker(0xc82001a180, 0xc82001a1e0)
    /home/zzz/go/src/github.com/zzz/asynch/main.go:55 +0x54
created by main.main
    /home/zzz/go/src/github.com/zzz/asynch/main.go:24 +0xf7

goroutine 7 [chan receive]:
main.worker(0xc82001a180, 0xc82001a1e0)
    /home/zzz/go/src/github.com/zzz/asynch/main.go:55 +0x54
created by main.main
    /home/zzz/go/src/github.com/zzz/asynch/main.go:24 +0xf7

goroutine 8 [chan receive]:
main.worker(0xc82001a180, 0xc82001a1e0)
    /home/zzz/go/src/github.com/zzz/asynch/main.go:55 +0x54
created by main.main
    /home/zzz/go/src/github.com/zzz/asynch/main.go:24 +0xf7

goroutine 9 [chan receive]:
main.worker(0xc82001a180, 0xc82001a1e0)
    /home/zzz/go/src/github.com/zzz/asynch/main.go:55 +0x54
created by main.main
    /home/zzz/go/src/github.com/zzz/asynch/main.go:24 +0xf7
exit status 2

代码:

package main

import (
    "log"
    "time"
)

type RowInfo struct {
    id int64
}

var QueueChan chan RowInfo

func main() {
    QueueChan := make(chan RowInfo)
    workerChan := make(chan RowInfo)
    exitChan := make(chan int64)

    go dispatcher(QueueChan, workerChan, exitChan)

    // Start WorkerCount number of workers
    workerCount := 4
    for i := 0; i < workerCount; i++ {
        go worker(workerChan, exitChan)
    }

    // Send test data
    for i := 0; i < 12; i++ {
        QueueChan <- RowInfo{id: int64(i)}
    }

    // Prevent app close
    for {
        time.Sleep(1 * time.Second)
    }
}

func dispatcher(queueChan, workerChan chan RowInfo, exitChan chan int64) {
    state := make(map[int64]bool)

    for {
        select {
        case job := <-QueueChan:
            if state[job.id] == true {
                continue
            }
            workerChan <- job
        case result := <-exitChan:
            state[result] = false
        }
    }
}

func worker(workerChan chan RowInfo, exitChan chan int64) {
    for job := range workerChan {
        log.Printf("Doing work on job rowInfo ID: %d", job.id)

        // Finish job
        exitChan <- job.id
    }
}

谢谢。

Answer 1

报错告诉你:所有的goroutine都在休眠，程序死锁了。

现在为什么你所有的 goroutines 都睡着了？让我们一一检查:

• worker goroutines:无限期等待 workerChan 上的新工作, 直到 workerChan 才会退出关闭，每当等待新工作时就睡着了
• dispatcher goroutine:永远循环，选择两个 channel 。永远不会退出，在select等待时睡着了
• main goroutine:在 time.Sleep 上永远循环, 永远不会退出，大部分时间都在 sleep

通常，在这种情况下，您会引入 chan struct{} (称之为 closing 或类似的名称)并将其包含在您的 select 中秒。如果你想关闭程序，只需 close(closing) . select会选择<-closing选项，你返回 goroutines。 您还应该添加 sync.WaitGroup当您的所有 goroutines 退出时收到通知。