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. workerpool Creates an unbuffered channel out to store results, but only returns after all results have been written to out. But since the read from the out channel occurs after out returns, and out 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 writing out Move the print statement after to see this)

Fix options include making out have a buffer of size n = number of results (i.e. out := make(chan int, n)) Or make out unbuffered and read from out while writing.

1. done The channel is also not being used correctly. Both main and workerpool 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 the case from the <code>workerpool and simply scope it by in as it is in Closed in main. 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!