Как я могу иметь одну функцию goroutine, которая ожидает значения от нескольких других? - PullRequest
Новая
       23

Как я могу иметь одну функцию goroutine, которая ожидает значения от нескольких других?

0 голосов
/ 16 февраля 2020

Я создал Go образец игровой площадки , который иллюстрирует то, о чем я говорю.

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

func func2(ch chan int) {
    fmt.Println("func2")
    v:=<-ch
    fmt.Println(v)
}

Затем в al oop я создаю подпрограммы для другой функции, и они являются частью WaitGroup. 

func func1(ch chan int, wg *sync.WaitGroup) {
    fmt.Println("func1")
    ch <- 11032
    wg.Done()
}

И в основном, я жду WaitGroup. Я захожу в тупик и не знаю, как это исправить. Чтобы было ясно, чего я пытаюсь достичь, я хочу, чтобы func2 оставался открытым как поток, как только я его вызвал, для обработки n значений, где n - количество вызываемых мной функций для func1. Я подумал об использовании WaitGroup Wait внутри func2, но я не хочу, чтобы он блокировался, поскольку он должен обрабатывать новые данные из func1 по мере их отправки.

Ответы [ 2 ]

1 голос
/ 17 февраля 2020

Я думаю, вы зашли в тупик, потому что ваш func2 потребляет только 1 значение из ch и затем завершает работу. Затем другие func1 goroutines застряли в ожидании готовности ch для записи, что они не могут сделать, потому что нет другой goroutine для чтения из ch на другом конце.

Так как вы Чтобы func2 непрерывно принимал значения от ch до закрытия ch, вам нужно создать al oop в func2 примерно так: 

func func2(ch chan int) {
    fmt.Println("func2")
    for v := range ch {
        fmt.Println(v)
    }
}

Это сохранит func2 " живы "и читают с ch до тех пор, пока вы не сделаете close(ch) где-то еще. Подходящее место для закрытия ch в вашем примере, вероятно, будет в main после wg.Wait().

Если вы хотите быть уверенным, что увидите результаты всех операторов Println до завершения программы, Вы также должны использовать некоторый механизм синхронизации, чтобы дождаться окончания func2 до sh. В противном случае main закончится сразу после close(ch), которое может или не может быть до того, как func2 напечатает каждое полученное значение.

Обычный метод для этого - "выполнено" канал. Например: 

func func2(ch chan int, done chan bool) {
    fmt.Println("func2")
    for v := range ch {
        fmt.Println(v)
    }
    done <- true
}

и main: 

done := make(chan bool)
go func2(ch, done)
...
wg.Wait()
close(ch)
<-done

Использование chan struct{} (пустая структура) также очень распространено, поскольку пустая структура не требует памяти.

0 голосов
/ 16 февраля 2020

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

var wg sync.WaitGroup

toUpdate := make(chan *someType, BufferSize)

for i := 0; i < BufferSize; i++ {
    go processEvents(toUpdate, &wg)
}

// // wait till all the checks have come back
go func(toUpdate chan * someType, group *sync.WaitGroup) {
    group.Wait()
    close(toCreate)
}(toCreate, toUpdate, &wg)
...