Multiple goroutines reading from the same channel

#php editor Strawberry will introduce to you the relevant content of multiple goroutines reading from the same channel in this article. In concurrent programming, goroutine is a lightweight thread in the Go language that can perform multiple tasks at the same time. Channels are an important way to communicate between goroutines. When multiple goroutines need to read data from the same channel, we need to pay attention to some issues and take corresponding measures to ensure the correctness and efficiency of the program. In what follows, we’ll explain the process in detail and provide some practical tips and advice.

Question content

Consider spawning multiple goroutines to read values from the same channel. The two workers are generated as expected, but only read one item from the channel and stop reading. I expect the goroutine to continue reading data from the channel until the goroutine sending the value to the channel is closed. Although something is preventing the sender from sending, the goroutine that spawned the project is not closed. Why does each worker only read one value and stop?

The output shows the two values sent, one read by each worker goroutine. The third value is sent but not read from either worker thread.

new worker new worker waiting sending 0 sending 1 sending 2 running func 1 sending value out 1 running func 0 sending value out 0

Go to the amusement park

package main import ( "fmt" "sync" ) func workerPool(done <-chan bool, in <-chan int, numberOfWorkers int, fn func(int) int) chan int { out := make(chan int) var wg sync.WaitGroup for i := 0; i < numberOfWorkers; i++ { fmt.Println("new worker") wg.Add(1) // fan out worker goroutines reading from in channel and // send output into out channel go func() { defer wg.Done() for { select { case <-done: fmt.Println("recieved done signal") return case data, ok := <-in: if !ok { fmt.Println("no more items") return } // fan-in job execution multiplexing results into the results channel fmt.Println("running func", data) value := fn(data) fmt.Println("sending value out", value) out <- value } } }() } fmt.Println("waiting") wg.Wait() fmt.Println("done waiting") close(out) return out } func main() { done := make(chan bool) defer close(done) in := make(chan int) go func() { for i := 0; i < 10; i++ { fmt.Println("sending", i) in <- i } close(in) }() out := workerPool(done, in, 2, func(i int) int { return i }) for { select { case o, ok := <-out: if !ok { continue } fmt.Println("output", o) case <-done: return default: } } }

Workaround

The previous comment about the channel not being buffered is correct, but there are other synchronization issues.

Unbuffered channels essentially mean that when a value is written, that value must be received before any other writes can occur.

1. workerpoolCreates an unbuffered channeloutto store results, but only returns after all results have been written to out. But since the read from the out channel occurs afteroutreturns, andoutis not buffered,workerpoolis blocked while trying to write, resulting in death Lock. That's why it looks like each worker is only sending a single value; in fact, after sending the first one, all workers are blocked because nothing can receive the value (you can do this by writingoutMove the print statement after to see this)

Fix options include makingouthave a buffer of sizen = number of results(i.e.out := make(chan int, n)) Or makeoutunbuffered and read fromoutwhile writing.

package main import ( "fmt" "sync" ) func workerPool(done chan bool, in <-chan int, numberOfWorkers int, fn func(int) int) chan int { out := make(chan int, 100) var wg sync.WaitGroup for i := 0; i < numberOfWorkers; i++ { fmt.Println("new worker") wg.Add(1) // fan out worker goroutines reading from in channel and // send output into out channel go func() { defer wg.Done() for data := range in { // fan-in job execution multiplexing results into the results channel fmt.Println("running func", data) value := fn(data) fmt.Println("sending value out", value) out <- value } fmt.Println("no more items") return }() } fmt.Println("waiting") wg.Wait() fmt.Println("done waiting") close(out) done <- true close(done) return out } func main() { done := make(chan bool, 1) in := make(chan int) go func() { for i := 0; i < 10; i++ { fmt.Println("sending", i) in <- i } close(in) }() out := workerPool(done, in, 2, func(i int) int { return i }) for { select { case o, ok := <-out: if !ok { continue } fmt.Println("output", o) case <-done: return } } }