Goroutines semaphores e pipeline pattern

O semáforo no Go é como se uma flag, que vai ser um channel que vai se comunicar com as routines, quando o semáforo apontar que deu tudo certo a ideia é q parem os canais.

func main() { channel := make(chan int) ok := make(chan bool) go func() { for i := 0; i < 10; i++ { channel <- i } ok <- true }() go func() { for i := 0; i < 10; i++ { channel <- i } ok <- true }() go func() { <-ok <-ok close(channel) }() for number := range channel { fmt.Println(number) } } 

Nesse exemplo a variável ok serve como o semáforo, perceba que na terceira função eu espero que a variável ok tenha sido esvaziada duas vezes para só então poder fechar o channel

Pipeline pattern

Até agora todas as vezes que usei o channel tinham sido de forma isolada, com a pipeline fazemos funções que retornam um channel e outras funções que recebem esse channel retornado e continuam o processamento. Aprendi a trabalhar com pipelines através da seguinte abordagem.

primeiro criei uma função generate que recebe numeros inteiros e retorna um channel de inteiros

func generate(numbers ...int) chan int { channel := make(chan int) go func() { for _, number := range numbers { channel <- number } }() return channel } 

E depois uma segunda função para dividir que recebe um channel de inteiros e retorna outro channel de inteiros

func divide(input chan int) chan int { channel := make(chan int) go func() { for number := range input { channel <- number / 2 } close(channel) }() return channel } 

porem no caso dessa função enquanto eu vou atribuindo o valor para o channel eu já atribuo ele dividido por 2.

Minha função main ficou da seguinte forma

func main() { numbers := generate(2, 4, 6) result := divide(numbers) fmt.Println(<-result) fmt.Println(<-result) fmt.Println(<-result) } 

Onde em ordem os prints exibem 1 2 3. Mas o que tá acontecendo de fato é. A função generate recebe o numero 2, atribui ele ao seu channel, e tenta atribuir o valor 4 em seguida, não consegue pois o channel ainda não foi esvaziado, nesse momento a função divide inicia e recebe o valor do channel da primeira função e atribui ao seu channel, ele também vai tentar atribuir o próximo numero e não vai conseguir, então ele chama o println para poder esvaziar o seu channel e isso roda 3 vezes até os 3 números terem sido processados

Fun in

Imagine que temos 3 processos rodando em paralelo cada um com o seu próprio canal, agora temos uma segunda função que recebe os canais dos 3 processos e retorna um novo canal mandando os resultados dos 3 processos

O fun in é um afunilamento que faz com que não precisemos olhar para os 3 channels e sim para apenas 1 e termos o resultado dos 3

func main() { x := funnel(generateMsg("Hello"), generateMsg("World")) for i := 0; i < 10; i++ { fmt.Println(<-x) } } func generateMsg(s string) <-chan string { channel := make(chan string) go func() { for i := 0; ; i++ { channel <- fmt.Sprintf("String %s - Value: %d", s, i) time.Sleep(time.Duration(rand.Intn(255)) * time.Millisecond) } }() return channel } func funnel(channel1, channel2 <-chan string) <-chan string { channel := make(chan string) go func() { for { channel <- <-channel1 } }() go func() { for { channel <- <-channel2 } }() return channel } 

ps: o operador <- <- me deixou um pouco confuso no inicio porem ele significa que estamos pegando o valor do channel1 e passando para a variavel channel

o output dessa função vai ser

❯ go run main.go String World - Value: 0 String Hello - Value: 0 String World - Value: 1 String Hello - Value: 1 String World - Value: 2 String World - Value: 3 String Hello - Value: 2 String Hello - Value: 3 String World - Value: 4 String World - Value: 5 

então somente olhando para a variável x dá para ter o resultado dos dois channels que estão rodando em paralelo

Fan out

Basicamente o Fan out é o processo oposto do fun in, onde vamos ter apenas 1 canal mas queremos distribuir ele pro diversos canais

para fazer o fan out podemos reutilizar o código do algoritmo de divide basta que pegarmos o valor inicial e passarmos para a função divide mais de uma vez, por exemplo:

func main() { c := generate(5, 10) d1 := divide(c) d2 := divide(c) fmt.Println(<-d1) fmt.Println(<-d2) } func generate(numbers ...int) chan int { channel := make(chan int) go func() { for _, n := range numbers { channel <- n } close(channel) }() return channel } func divide(input chan int) chan int { channel := make(chan int) go func() { for number := range input { channel <- number / 2 } close(channel) }() return channel }