Как правильно использовать каналы для управления параллелизмом?

Question

Я новичок в параллелизме в Go и пытаюсь понять, как использовать каналы для управления параллелизмом. То, что я хотел бы сделать, - это цикл, в котором я могу вызывать функцию, используя новую подпрограмму go, и продолжать цикл, пока эта функция обрабатывается, и я хотел бы ограничить число выполняемых подпрограмм до 3. Моя первая попытка сделать это был код ниже:

func write(val int, ch chan bool) {
    fmt.Println("Processing:", val)
    time.Sleep(2 * time.Second)
    ch <- val % 3 == 0
}

func main() {
    ch := make(chan bool, 3) // limit to 3 routines?
    for i := 0; i< 10; i++ {
        go write(i, ch)
        resp := <- ch
        fmt.Println("Divisible by 3:", resp)
    }
    time.Sleep(20 * time.Second)
}

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

Что мне нужно изменить в этом примере, чтобы получить функциональность, описанную выше?

Answer 1

Существует еще один способ запустить процедуры go параллельно с ожиданием, пока все они вернут значение в канале, используя Wait groups.Это также помогает синхронизировать подпрограммы.Если вы работаете с подпрограммами go, чтобы дождаться завершения всех из них перед выполнением другой функции, лучше использовать wait group.

package main

import (
    "fmt"
    "time"
    "sync"
)

func write(val int, wg *sync.WaitGroup, ch chan bool) {
    defer wg.Done()
    fmt.Println("Processing:", val)
    time.Sleep(2 * time.Second)
    ch <- val % 3 == 0
}
func main() {
    wg := &sync.WaitGroup{}
    ch := make(chan bool, 3)
    for i := 0; i< 10; i++ {
        wg.Add(1)
        go write(i, wg, ch)
    }
    for i := 0; i< 10; i++ {
        fmt.Println("Divisible by 3: ", <-ch)
    }
    close(ch)
    wg.Wait()
    time.Sleep(20 * time.Second)
}

Пример игровой площадки

Answer 2

Проблема здесь очень проста:

for i := 0; i< 10; i++ {
    go write(i, ch)
    resp := <- ch
    fmt.Println("Divisible by 3:", resp)
}

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

Чтобы это исправить, вам нужно раскрутить все 10 горутин, затем дождаться всех 10 ответов ( детская площадка ):

for i := 0; i< 10; i++ {
    go write(i, ch)
}
for i := 0; i<10; i++ {
    resp := <- ch
    fmt.Println("Divisible by 3:", resp)
}

Теперь у вас на канале заблокировано 7 процедур, но оно настолько короткое, что вы не можете видеть, что это происходит, поэтому вывод не будет очень интересным. Если вы попробуете добавить сообщение Processed в конце программы и спать между каждым прочитанным каналом, вы увидите, что 3 из них завершают работу немедленно (ну, после ожидания 2 секунд), а затем другие разблокируют и заканчивают одно по одному ( детская площадка ).