Закрытие канала неизвестной длины

Question

Я не в состоянии закрыть канал, когда нет знания о его
длина

package main 

import (
    "fmt" 
    "time" 
) 

func gen(ch chan int) { 
    var i int 
    for { 
     time.Sleep(time.Millisecond * 10) 
     ch <- i 
     i++ 
     // when no more data (e.g. from db, or event stream) 
     if i > 100 { 
      break 
     } 
    } 

    // hot to close it properly? 
    close(ch) 
} 

func receiver(ch chan int) { 
    for i := range ch { 
     fmt.Println("received:", i) 
    } 
} 

func main() { 
    ch := make(chan int) 

    for i := 0; i < 10; i++ { 
     go gen(ch) 
    } 

    receiver(ch) 
} 

Это дает мне ошибку

panic: send on closed channel 

goroutine 8 [running]: 
main.gen(0xc82001a0c0) 
    /home/exu/src/github.com/exu/go-workshops/100-concurrency-channels/16-close-problem.go:12 +0x57 
created by main.main 
    /home/exu/src/github.com/exu/go-workshops/100-concurrency-channels/16-close-problem.go:35 +0xbd 

goroutine 1 [panicwait]: 
runtime.gopark(0x0, 0x0, 0x50b8e0, 0x9, 0x10, 0x1) 
    /usr/lib/go/src/runtime/proc.go:185 +0x163 
runtime.main() 
    /usr/lib/go/src/runtime/proc.go:121 +0x2f4 
runtime.goexit() 
    /usr/lib/go/src/runtime/asm_amd64.s:1696 +0x1 

goroutine 6 [sleep]: 
time.Sleep(0x989680) 
    /usr/lib/go/src/runtime/time.go:59 +0xf9 
main.gen(0xc82001a0c0) 
    /home/exu/src/github.com/exu/go-workshops/100-concurrency-channels/16-close-problem.go:11 +0x29 
created by main.main 
    /home/exu/src/github.com/exu/go-workshops/100-concurrency-channels/16-close-problem.go:33 +0x79 

goroutine 7 [sleep]: 
time.Sleep(0x989680) 
    /usr/lib/go/src/runtime/time.go:59 +0xf9 
main.gen(0xc82001a0c0) 
    /home/exu/src/github.com/exu/go-workshops/100-concurrency-channels/16-close-problem.go:11 +0x29 
created by main.main 
    /home/exu/src/github.com/exu/go-workshops/100-concurrency-channels/16-close-problem.go:34 +0x9b 
exit status 2 

Это логично - закрытие первого goroutine канал, когда второй пытается отправить его. Каким будет наилучший подход к закрытию канала в этой ситуации?

Answer 1

Как только канал закрыт, вы не можете отправлять дополнительные значения, иначе он паникует. Это то, что вы испытываете.

Это потому, что вы запускаете несколько goroutines, которые используют один и тот же канал, и отправляют на него значения. И вы закрываете канал в каждом из них. И поскольку они не синхронизированы, как только первая горутин достигает точки, где она закрывает ее, другие могут (и они будут) продолжать отправлять значения на нее: panic!

Вы можете закрыть канал только один раз (попытка закрытия уже закрытого канала также вызывает панику). И вы должны сделать это, когда все горуты, которые отправляют значения на нем, выполняются. Для этого вам нужно определить, когда все отправители отправляются. Идиоматический способ обнаружить это - использовать sync.WaitGroup.

Для каждого начатого отправителя goroutine добавьте 1 к WaitGroup, используя WaitGroup.Add(). И каждый горутин, который совершает отправку значений, может сигнализировать об этом, вызывая WaitGroup.Done(). Лучше всего сделать это в качестве отложенного заявления, поэтому, если ваш горутин резко прекратится (например, паники), WaitGroup.Done() все равно будет называться, и он не оставит свисающих других горутов (ожидая отпущения греха - «отсутствующий» звонок WaitGroup.Done(), который никогда не появлялся ...).

И WaitGroup.Wait() будет ждать завершения всех отправителей goroutines, и только после этого и только один раз он закроет канал. Мы хотим обнаружить это «глобальное» событие и закрыть канал, в то время как обработка отправленных на него значений выполняется, поэтому мы должны сделать это в своем собственном goroutine.

Приемник goroutine будет работать до тех пор, пока канал не будет закрыт, так как мы использовали конструкцию for ... range на канале. И поскольку он запускается в главном goroutine, программа не выйдет, пока все значения не будут правильно приняты и обработаны из канала. Конструкция контуров for ... range до тех пор, пока все значения не будут получены, которые были отправлены до закрытия канала.

Обратите внимание, что решение ниже работает с буферизованным и небуферизованным каналом без изменений (попробуйте использовать буферный канал с ch := make(chan int, 100)).

Правильное решение (попробуйте на Go Playground):

func gen(ch chan int, wg *sync.WaitGroup) { 
    defer wg.Done() 
    var i int 
    for { 
     time.Sleep(time.Millisecond * 10) 
     ch <- i 
     i++ 
     // when no more data (e.g. from db, or event stream) 
     if i > 100 { 
      break 
     } 
    } 
} 

func receiver(ch chan int) { 
    for i := range ch { 
     fmt.Println("received:", i) 
    } 
} 

func main() { 
    ch := make(chan int) 
    wg := &sync.WaitGroup{} 

    for i := 0; i < 10; i++ { 
     wg.Add(1) 
     go gen(ch, wg) 
    } 

    go func() { 
     wg.Wait() 
     close(ch) 
    }() 

    receiver(ch) 
} 

Примечание:

Обратите внимание, что это важно, что receiver(ch) работает в главном goroutine, и код, что ждет WaitGroup и закрывает канал в своей (не основной) горутине; а не наоборот. Если вы переключите эти 2, это может привести к «раннему завершению», то есть не все значения могут быть получены и обработаны из канала. Причина этого в том, что программа Go выходит, когда заканчивается главный горутин (спецификация: Program execution). Это не дожидается завершения других (не главных) goroutines.Поэтому, если ожидание и закрытие канала будет в главной горутине, после закрытия канала программа может выйти в любой момент, не дожидаясь, пока другой горутин, который в этом случае будет зацикливаться, чтобы получать значения из канала.

Answer 2

использовать, кроме как для выбора.

for{ 
    i,ok:=<-ch 
    //process with i 
    if !ok { 
     break 
    } 
}