Как читать с одного канала, совместно используемого несколькими программами

Question

У меня есть несколько goroutines, пишущих на тот же канал.Если я использую буферизованный канал, я могу получить вход.Однако, если используется небуферизованный канал, я могу прочитать только половину значений:

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()
}

Это функция получателя:

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

Функция отправителя:

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

И вывод:

testAsyncFunc 2 0 4 6 8 2 Готово ...

Пока я ожидаю получить 10 номеров.

Answer 1

Выбор по умолчанию не работает так, как вы думаете.Если блок выбора имеет регистр по умолчанию, он выберет его немедленно , если ни один из других регистров не готов к чтению.

Это означает, что ваш приемник3 с большой вероятностью попадет в свой регистр по умолчанию.

Answer 2

Код вернется с ошибкой, если вы не создадите буферные каналы по причине того, что канал закрыт до того, как все значения отправлены на него.

Не закрывайте каналы и дождитесь завершения процедур go.Если вы хотите закрыть каналы, закройте, когда все значения, отправленные по каналу, будут получены внутри подпрограммы приема приемника.

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()

Еще одна вещь, которую вы заметили, вы увеличили счетчик группы ожидания до 10, пока выуменьшение счетчика после запуска подпрограммы go внутри цикла for, что неправильно.Вы должны уменьшить счетчик группы ожидания внутри подпрограммы отправителя, когда она завершит свое выполнение.

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

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

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
        }
    }
}

Создайте подпрограмму go, чтобы закрыть канал и дождаться завершения подпрограмм отправителя go.Поэтому нижеприведенный код будет ожидать завершения всех ваших подпрограмм go для отправки значений по каналам, а затем закроет канал:

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
}

Выход

testAsyncFunc2
9
0
1
2
3
4
5
6
7
8

Рабочий код на игровая площадка Go