如何从多个goroutine共享的单个通道读取

Question

I have several goroutines writing to the same channel. If I use a buffered channel I can retrieve the input. However if use an unbuffered channel I can only read about half the values:

func testAsyncFunc2() {

    ch := make(chan int,10)

    fmt.Println("testAsyncFunc2")
    wg.Add(10)
    for  i :=0; i < 10; i++  {
        go sender3(ch, i)
        wg.Done()
    }

    receiver3(ch)

    close(ch)
    wg.Wait()
}

This is the receiver function:

func receiver3(ch chan int) {
    for {
        select {
        case <-ch:
            fmt.Println(<-ch)
        default:
            fmt.Println("Done...")
            return
        }
    }
}

Sender function:

func sender3(ch chan int, i int) {
    ch <- i
}

And the output:

testAsyncFunc 2 0 4 6 8 2 Done...

While I would expect to get back 10 numbers.

Answer 1

The code will return with an error if you not create buffer channels reason being the channel is closed before all the values sent on it.

Do not close the channels and wait for the go routines to complete. If you want to close the channels, close when all the values send on the channel are received inside the receiver go routine.

fmt.Println("testAsyncFunc2")
for  i :=0; i < 10; i++  {
    wg.Add(1)
    go sender3(ch, i)
}

receiver3(ch)
close(ch) // this will close the channels before all the values sent on it will be received.
wg.Wait()

One more thing to notice you have increased wait group counter to 10 while you are decreasing the counter after you launch the go routine inside the for loop which is wrong. You should decrease the wait group counter inside the sender go routine when it is finished its execution.

func sender(ch chan int, int){
     defer wg.Done()
}

In the for loop the select default condition will run before all the values sent on the channel is received that's why all values send will not print. Since the loop will return when there is no value send on the channel.

func receiver3(ch chan int) {
    for {
        select {
        case <-ch:
            fmt.Println(<-ch)
        default: // this condition will run when value is not available on the channel.
            fmt.Println("Done...")
            return
        }
    }
}

Create a go routine to close the channel and wait for the sender go routines to finish. So below code will wait for your all go routines to finish sending the values on channels and then it will close the channel:

package main

import (
    "fmt"
    "sync"
)

var wg sync.WaitGroup

func main() {
    ch := make(chan int)
    fmt.Println("testAsyncFunc2")
    for i := 0; i < 10; i++ {
        wg.Add(1)
        go sender(ch, i)
    }
    receiver3(ch)
    go func() {
        defer close(ch)
        wg.Wait()
    }()
}

func receiver3(ch <-chan int) {
    for i := 0; i < 10; i++ {
        select {
        case value, ok := <-ch:
            if !ok {
                ch = nil
            }
            fmt.Println(value)
        }
        if ch == nil {
            break
        }
    }
}

func sender(ch chan int, i int) {
    defer wg.Done()
    ch <- i
}

Output

testAsyncFunc2
9
0
1
2
3
4
5
6
7
8

Working Code on Go playground