我需要你的幫助來理解為什么我的readFromWorkerfunc 會(huì)導(dǎo)致死鎖。當(dāng)我注釋掉下面這樣的行時(shí)，它可以正常工作（因此我知道問題就在這里）。整個(gè)在這里https://play.golang.org/p/-0mRDAeD2tr我將衷心感謝您的幫助func readFromWorker(inCh <-chan *data, wg *sync.WaitGroup) { defer func() { wg.Done() }() //stageIn1 := make(chan *data) //stageOut1 := make(chan *data) for v := range inCh { fmt.Println("v", v) //stageIn1 <- v } //go stage1(stageIn1, stageOut1) //go stage2(stageOut1)}
1 回答
胡子哥哥
TA貢獻(xiàn)1825條經(jīng)驗(yàn) 獲得超6個(gè)贊
我已經(jīng)評(píng)論了你做錯(cuò)的相關(guān)部分。另外，我建議考慮一個(gè)更好的模式。
請(qǐng)記住，for range頻道不會(huì)停止循環(huán)，除非close它正在循環(huán)的同一頻道被調(diào)用。此外，關(guān)閉通道的經(jīng)驗(yàn)法則是發(fā)送到通道的發(fā)送者也必須關(guān)閉它，因?yàn)榘l(fā)送到關(guān)閉的通道會(huì)導(dǎo)致panic.
此外，在使用無緩沖和緩沖通道時(shí)要非常小心。對(duì)于無緩沖通道，發(fā)送方和接收方必須準(zhǔn)備好，否則在您的情況下也會(huì)發(fā)生死鎖。
package main
import (
"fmt"
"sync"
)
type data struct {
id int
url string
field int
}
type job struct {
id int
url string
}
func sendToWorker(id int, inCh <-chan job, outCh chan<- *data, wg *sync.WaitGroup) {
// wg.Done() is itself a function call, no need to wrap it inside
// an anonymous function just to use defer.
defer wg.Done()
for v := range inCh {
// some pre process stuff and then pass to pipeline
outCh <- &data{id: v.id, url: v.url}
}
}
func readFromWorker(inCh <-chan *data, wg *sync.WaitGroup) {
// wg.Done() is itself a function call, no need to wrap it inside
// an anonymous function just to use defer.
defer wg.Done()
var (
stageIn1 = make(chan *data)
stageOut1 = make(chan *data)
)
// Spawn the goroutines so that there's no deadlock
// as the sender and receiver both should be ready
// when using unbuffered channels.
go stage1(stageIn1, stageOut1)
go stage2(stageOut1)
for v := range inCh {
fmt.Println("v", v)
stageIn1 <- v
}
close(stageIn1)
}
func stage1(in <-chan *data, out chan<- *data) {
for s := range in {
fmt.Println("stage1 = ", s)
out <- s
}
// Close the out channel
close(out)
}
func stage2(out <-chan *data) {
// Loop until close
for s := range out {
fmt.Println("stage2 = ", s)
}
}
func main() {
const chanBuffer = 1
var (
inputsCh = make(chan job, chanBuffer)
resultsCh = make(chan *data, chanBuffer)
wgInput sync.WaitGroup
wgResult sync.WaitGroup
)
for i := 1; i <= 4; i++ {
wgInput.Add(1)
go sendToWorker(i, inputsCh, resultsCh, &wgInput)
}
wgResult.Add(1)
go readFromWorker(resultsCh, &wgResult)
for j := 1; j <= 10; j++ {
inputsCh <- job{id: j, url: "google.com"}
}
close(inputsCh)
wgInput.Wait()
close(resultsCh)
wgResult.Wait()
}
- 1 回答
- 0 關(guān)注
- 130 瀏覽
添加回答
舉報(bào)
0/150
提交
取消