Каналы и конкуренция в Go: Танец горутин в Golang!

GoRoutines

Итак, GoRoutine – это функция, которая выполняется одновременно с другими goroutines. Мы используем goroutines для запуска нескольких потоков, что помогает нам достичь параллелизма.

Конкуренция – это способность программы выполнять несколько задач независимо друг от друга в течение нескольких периодов времени.

В чем разница между параллелизмом и конкуренцией?

func randSleep(name String, limit int, sleep int){
 for(i=1;i<=n;i++){
 fmt.println(name,rand.Intn(i))
 time.sleep(time.Duration(sleep*int(time.second)))
 }
}

func main(){
go randSleep("first:",4,3)
go randSleep("second:",4,3)
}

Приведенная выше программа ничего не выведет на терминал, потому что главная функция завершается до того, как выполняются гороутины.

Но если после горутины идет последовательный код, то пока эта горутина будет выполняться, она будет выполняться до выполнения последовательного кода. Например:

func randSleep(name String, limit int, sleep int){
 for i=1;i<=n;i++{
 fmt.println(name,rand.Intn(i))
 time.sleep(time.Duration(sleep*int(time.second)))
 }
}

func main(){
 go randSleep("first:",4,3)
 randSleep("second:",4,3)
}

Теперь, в приведенном выше коде, goroutine будет выполняться одновременно с последовательным кодом, независимо от того, завершена goroutine или нет, потому что она ожидает выполнения нашего последовательного кода.

Чтобы решить эту проблему, мы будем использовать группы ожидания, которые являются частью пакета sync.

WaitGroup

Итак, WaitGroup позволяют программе ожидать выполнения заданных горутин. Это механизмы синхронизации в Golang, которые блокируют выполнение программы до тех пор, пока горутины в WaitGroup полностью не выполнятся.

func randSleep(wg *sync.WaitGroup, name String, limit int, sleep int){

defer wg.Done() //altering/subtracting the counter of goroutine when it completes 
 for i:=1;i<=limit;i++{
 fmt.println(name,rand.Intn(i))
 }
}

func main(){
wg:=new(sync.WaitGroup) //creates a new wait group 

wg.Add(2) //it informs that it must wait for two goroutines

go randSleep(wg, "first:",5,3)
go randSleep(wg,"second:",5,3)

wg.Wait() //block the execution until the goroutine's execution completes. 
}

Таким образом, весь процесс выглядит как прибавление к счетчику в wg.Add(), вычитание из него в wg.Done() и ожидание, когда счетчик достигнет 0 в wg.Wait().

Каналы

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

Двунаправленные каналы в Go блокируются, то есть при отправке данных в канал Go ждет, пока данные не будут считаны из канала, прежде чем продолжить выполнение.

func writeChannel(wg *sync.WaitGroup, ch chan int, stop int){
defer wg.Done()
for i:=1;i<=stop;i++{
ch<-i
}
}

func readChannel(wg *sync.WaitGroup, ch chan int, stop int){
defer wg.Done()
for i:=1;i<=stop;i++{
fmt.println(<-ch)
}
}

func main(){
wg:=new(sync.WaitGroup)
wg.Add(2)
limitChannel:=make(chan int) //unbuffered channel
defer close(limitChannel)
go writeChannel(wg,limitChannel,3)
go readChannel(wg,limitChannel,3)
wg.Wait()
}

Таким образом, количество операций отправки должно быть равно количеству операций получения, иначе мы получим тупик.

Как в приведенном выше коде может возникнуть ошибка?

Если в горутину writeChannel записывается больше значений в канал, чем может прочитать горутина read Channel, то при попытке записи горутина writeChannel в конце концов заблокируется, поскольку буфер канала переполнен.

Поскольку горутина writeChannel заблокирована и не может продолжить выполнение, она не может достичь оператора wg.Done(), который уменьшил бы счетчик sync.WaitGroup.

В результате вызов wg.Wait() в функции main никогда не вернется, и программа окажется в состоянии тупика.

Типы каналов в go

Go в основном состоит из двух типов каналов: буферизованных и небуферизованных. Мы рассмотрим, чем они отличаются друг от друга и в чем их преимущества.

Буферизованные каналы

1. Мы можем создать канал, хранящий несколько значений, и тогда отправка данных в канал не будет блокироваться до тех пор, пока мы не превысим его емкость.
2. В буферизованных каналах обмен данными между горутинами происходит асинхронно. Асинхронность в основном означает, что наша программа не будет находиться в состоянии блока, а остальная ее функциональность будет выполняться.
3. Буферизованный канал повторяет структуру данных очереди, то есть значения, вставленные первыми в канал, будут выводиться первыми.
4. В случае буферизованных каналов горутина отправителя не будет заблокирована, даже если мы заполним значение в канале. Он отправит данные в канал, и эти данные, по сути, будут поставлены в очередь, а горутина продолжит делать то, что должна была.

func writeChannel(wg *sync.WaitGroup, limitchannel chan int, stop int) {
defer wg.Done()
for i := 1; i <= stop; i++ {
limitchannel <- i
}
}

func main() {
wg := new(sync.WaitGroup)
wg.Add(1)
limitchannel:= make(chan int, 2)
defer close(limitchannel)
go writeChannel(wg, limitchannel, 2)
wg.Wait()
fmt.Println(<-limitchannel)
fmt.Println(<-limitchannel)
}

Что делать, если буферизованный канал достиг своей пропускной способности?

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

Небуферизованные каналы

1. По умолчанию каналы являются небуферизованными, т. е. им нужен получатель, как только в канал будет отправлено сообщение.
2. Горутина отправителя будет находиться в состоянии блока, пока горутина получателя не прочитает сообщение из канала.
3. По умолчанию горутина отправителя переходит в состояние ожидания, как только мы загружаем данные в канал.
4. Небуферизованные каналы позволяют горутинам взаимодействовать только синхронно. Синхронное взаимодействие – это когда отправитель должен ожидать ответа от получателя.

Небуферизованные каналы: Модель Fork-Join

func main(){
myChannel:=make(chan string)

//This is my anonymous function
go func(){
myChannel <- "data"
}() //here I'm invoking the function

msg:=<-myChannel
fmt.Println(msg)
}

В приведенном выше примере процедура go, созданная main, была форкетирована (освобождена от main) и что-то делает. Но мы снова подключаемся к ней с помощью канала.

Таким образом, main не завершится, пока мы не получим значение из канала или не закроем канал.

select

Оператор select блокирует выполнение функции main до тех пор, пока не будет запущен один из ее случаев, и, получив сообщение от канала, он выполнит код в этом блоке.

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

func main(){

go func(){
myChannel <- "data"
}()

go func(){
anotherChannel <- "data data"
}()

select{
case msgFromMyChannel := <-myChannel:
fmt.Println(msgFromMyChannel)

case msgFromAnotherChannel := <-anotherChannel:
fmt.Println(msgFromAnotherChannel)
}
}

func main(){

charChannel:= make(chan string,3)
chars:=[]string{"a","b","c"}

for _,s:=range chars{
select{
case charChannel <- s:
  }
}

close(charChannel)
for result:= range charChannel{
fmt.Println(result)
  }
}

Закрыть каналы

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

Как только мы закроем канал, он больше не будет принимать значения и отправит сообщение о том, что канал закрыт.

func doWork(done <-chan bool){ //this way, the channel will behave as a read-only channel.
for{

Select{
case <- done:
return
default:
fmt.Println("Doing Work")
    }
  }
}

func main(){
done:= make(chan bool)
go doWork(done)
time.Sleep(time.second*3)
close(done) // The main go routine will send a signal after 3 seconds to its child goroutine, and by that way it will terminate the child goroutine.
}

Примечание: Как говорится в The Go Programming Language, нет необходимости закрывать каждый канал, когда вы закончили работу с ним. Закрывать канал нужно только тогда, когда важно сообщить принимающим горутинам, что все данные отправлены.

Пайплайн

Согласно определению, это всего лишь один из многих видов паттернов параллельного проектирования. Неформально конвейер – это группа этапов, соединенных каналами и обрабатываемых горутинами.

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

начало (вход) -> этап 1 (сделать что-то) -> этап 2 (сделать что-то) -> конец

Вышеописанная модель называется “конвейером”, в котором каждый этап отвечает за выполнение некоторой операции, а мы продолжаем передавать данные от одного этапа к другому.

func sliceToChannel(nums []int) <-chan int{ //it will return read-only channel
out:= make(chan int)
go func(){
for _,n := range nums{
out <- n
    }
close(out)
  }()
return out
}

• Пояснение к приведенной выше функции: Итак, мы создали небуферизованный канал, что означает, что он будет заблокирован до тех пор, пока мы не прочитаем значение из канала. Поэтому, как только мы вставим значение в наш канал внутри цикла, этот цикл будет заблокирован до тех пор, пока мы не прочитаем значение. Таким образом, оператор return будет выполнен, а в фоновом режиме горутина все еще будет работать.

func sq(in <-chan int) <-chan int{
out:= make(chan int)
go func(){
for n:= range in{
out <- n*n
    }
close(out)
  }()
return out
}

• Объяснение вышеприведенной функции: В этой функции мы выполняем цикл по каналу, который мы вернули из функции sliceToChannel. Таким образом, наш канал содержит значение, которое мы будем считывать, и как только мы закончим считывание, следующее значение будет вставлено в канал через горутину, которая выполняется в фоновом режиме функции sliceToChannel. Таким образом, обе функции будут работать синхронно.

func main(){
//input
nums:=[]int{2,3,4,7,1}
//stage 1
dataChannel:= sliceToChannel(nums)
//stage 2
finalChannel := sq(dataChannel)
//stage 3
for n:=range finalChannel{
fmt.Println(n)
  }
}

Спасибо, что дочитали до конца.