給定 Go 中（部分）填充的緩沖通道ch := make(chan *MassiveStruct, n)for i := 0; i < n; i++ { ch <- NewMassiveStruct()}是否建議在關(guān)閉通道時(shí)（由作者）也排空通道，以防讀者何時(shí)從中讀取（例如，數(shù)量有限并且他們目前正忙）？那是close(ch)for range ch {}如果通道上還有其他并發(fā)閱讀器，這樣的循環(huán)是否保證結(jié)束？上下文：具有固定數(shù)量工作器的隊(duì)列服務(wù)，當(dāng)服務(wù)關(guān)閉時(shí)，它應(yīng)該停止處理任何排隊(duì)的東西（但不一定在之后被 GC 處理）。所以我要向工作人員表明服務(wù)正在終止。我可以立即排空剩余的“隊(duì)列”，讓 GC 釋放分配的資源，我可以讀取和忽略工作程序中的值，我可以離開通道，因?yàn)樽x取器正在運(yùn)行，并將通道設(shè)置為 nil 在寫入器中GC 清理一切。我不確定哪種方式最干凈。
3 回答
FFIVE
TA貢獻(xiàn)1797條經(jīng)驗(yàn) 獲得超6個(gè)贊
這取決于您的程序，但一般來說我傾向于說不（您不需要在關(guān)閉頻道之前清除頻道）：如果您關(guān)閉頻道時(shí)頻道中有項(xiàng)目，任何仍在從頻道閱讀的讀者都會(huì)接收項(xiàng)目，直到通道為空。
下面是一個(gè)例子：
package main
import (
"sync"
"time"
)
func main() {
var ch = make(chan int, 5)
var wg sync.WaitGroup
wg.Add(1)
for range make([]struct{}, 2) {
go func() {
for i := range ch {
wg.Wait()
println(i)
}
}()
}
for i := 0; i < 5; i++ {
ch <- i
}
close(ch)
wg.Done()
time.Sleep(1 * time.Second)
}
在這里，程序?qū)⑤敵鏊许?xiàng)目，盡管通道在任何讀者甚至可以從通道讀取之前嚴(yán)格關(guān)閉。
嚕嚕噠
TA貢獻(xiàn)1784條經(jīng)驗(yàn) 獲得超7個(gè)贊
有更好的方法來實(shí)現(xiàn)您想要實(shí)現(xiàn)的目標(biāo)。您當(dāng)前的方法只會(huì)導(dǎo)致丟棄一些記錄，并隨機(jī)處理其他記錄（因?yàn)榕潘h(huán)正在與所有消費(fèi)者競(jìng)爭(zhēng)）。這并沒有真正解決目標(biāo)。
你想要的是取消。這是Go 并發(fā)模式中的一個(gè)示例：管道和取消
func sq(done <-chan struct{}, in <-chan int) <-chan int {
out := make(chan int)
go func() {
defer close(out)
for n := range in {
select {
case out <- n * n:
case <-done:
return
}
}
}()
return out
}
您將一個(gè)done通道傳遞給所有 goroutine，并在您希望它們都停止處理時(shí)關(guān)閉它。如果你經(jīng)常這樣做，你可能會(huì)發(fā)現(xiàn)這個(gè)golang.org/x/net/context包很有用，它正式化了這個(gè)模式，并添加了一些額外的功能（比如超時(shí)）。
動(dòng)漫人物
TA貢獻(xiàn)1815條經(jīng)驗(yàn) 獲得超10個(gè)贊
我覺得所提供的答案實(shí)際上并沒有澄清，除了不需要排水也不需要關(guān)閉的提示。因此，針對(duì)所描述的上下文的以下解決方案對(duì)我來說看起來很干凈，它終止了工作人員并刪除了對(duì)他們或相關(guān)頻道的所有引用，因此，讓 GC 清理頻道及其內(nèi)容：
type worker struct {
submitted chan Task
stop chan bool
p *Processor
}
// executed in a goroutine
func (w *worker) run() {
for {
select {
case task := <-w.submitted:
if err := task.Execute(w.p); err != nil {
logger.Error(err.Error())
}
case <-w.stop:
logger.Warn("Worker stopped")
return
}
}
}
func (p *Processor) Stop() {
if atomic.CompareAndSwapInt32(&p.status, running, stopped) {
for _, w := range p.workers {
w.stop <- true
}
// GC all workers as soon as goroutines stop
p.workers = nil
// GC all published data when workers terminate
p.submitted = nil
// no need to do the following above:
// close(p.submitted)
// for range p.submitted {}
}
}
- 3 回答
- 0 關(guān)注
- 162 瀏覽
添加回答
舉報(bào)
0/150
提交
取消