#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.
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: } } }
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.
workerpool
Creates an unbuffered channelout
to store results, but only returns after all results have been written to out. But since the read from the out channel occurs afterout
returns, andout
is not buffered,workerpool
is 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 writingout
Move the print statement after to see this)Fix options include makingout
have a buffer of sizen = number of results
(i.e.out := make(chan int, n)
) Or makeout
unbuffered and read fromout
while writing.
done
The channel is also not being used correctly. Bothmain
andworkerpool
rely on it to stop execution, but nothing is written to it! It is also unbuffered and therefore suffers from the deadlock problem mentioned above.To fix this you can first remove thecase <-done:
from theworkerpool
and simply scope it byin
as it is in Closed inmain
.done
can then be set to a buffered channel to resolve the deadlock.
Combining these fixes results in:
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 } } }
This may solve your problem, but it's not the best way to use channels! The structure itself can be changed simpler without having to rely on buffered channels.
The above is the detailed content of Multiple goroutines reading from the same channel. For more information, please follow other related articles on the PHP Chinese website!