Go Starter Pack de Concorrência

Visão geral

Breve introdução

A linguagem Go possui diversas ferramentas para gerenciamento de concorrência. Neste artigo, apresentaremos algumas delas e seus truques.

Goroutines?

Goroutine é um novo modelo de concorrência suportado pela linguagem Go. Normalmente, um programa recebe threads do sistema operacional para executar várias tarefas simultaneamente e realiza operações em paralelo, de acordo com o número de núcleos. Para realizar concorrência em unidades menores, threads verdes são criadas no espaço do usuário, permitindo que várias threads verdes operem dentro de uma thread do sistema operacional. No entanto, as goroutines tornaram essas threads verdes ainda menores e mais eficientes. Essas goroutines usam menos memória do que threads e podem ser criadas e alternadas mais rapidamente do que threads.

Para usar goroutines, basta usar a palavra-chave go. Isso permite que o código síncrono seja executado como código assíncrono de forma intuitiva durante o processo de escrita do programa.

 1package main
 2
 3import (
 4    "fmt"
 5    "time"
 6)
 7
 8func main() {
 9    ch := make(chan struct{})
10    go func() {
11        defer close(ch)
12        time.Sleep(1 * time.Second)
13        fmt.Println("Olá, Mundo!")
14    }()
15
16    fmt.Println("Aguardando goroutine...")
17    for range ch {}
18}

Este código simplesmente altera o código síncrono, que imprime Olá, Mundo! após uma pausa de 1 segundo, para um fluxo assíncrono. O exemplo atual é simples, mas se um código um pouco mais complexo for alterado de síncrono para assíncrono, a legibilidade, visibilidade e compreensão do código se tornam ainda melhores do que abordagens como async await ou promise.

No entanto, em muitos casos, se um fluxo que simplesmente chama esse código síncrono de forma assíncrona e um fluxo como fork & join (um fluxo semelhante a dividir para conquistar) não forem compreendidos, um código ruim de goroutine pode ser criado. Neste artigo, apresentaremos alguns métodos e técnicas para lidar com esses casos.

Gerenciamento de Concorrência

Contexto

Pode ser surpreendente que o context apareça como a primeira técnica de gerenciamento. No entanto, em Go, o context desempenha um papel excelente no gerenciamento de toda a árvore de trabalho, além de uma simples função de cancelamento. Para aqueles que não conhecem, vamos explicar brevemente este pacote.

 1package main
 2
 3func main() {
 4    ctx, cancel := context.WithCancel(context.Background())
 5    defer cancel()
 6
 7    go func() {
 8        <-ctx.Done()
 9        fmt.Println("Contexto concluído!")
10    }()
11
12    time.Sleep(1 * time.Second)
13
14    cancel()
15
16    time.Sleep(1 * time.Second)
17}

O código acima usa context para imprimir Contexto concluído! após 1 segundo. context pode verificar o cancelamento por meio do método Done(), e fornece vários métodos de cancelamento, como WithCancel, WithTimeout, WithDeadline e WithValue.

Vamos criar um exemplo simples. Suponha que você esteja escrevendo um código que usa o padrão aggregator para buscar user, post e comment para obter alguns dados. E se todas as solicitações tiverem que ser concluídas em 2 segundos, você pode escrever da seguinte forma:

 1package main
 2
 3func main() {
 4    ctx, cancel := context.WithTimeout(context.Background(), 2 * time.Second)
 5    defer cancel()
 6
 7    ch := make(chan struct{})
 8    go func() {
 9        defer close(ch)
10        user := getUser(ctx)
11        post := getPost(ctx)
12        comment := getComment(ctx)
13
14        fmt.Println(user, post, comment)
15    }()
16
17    select {
18    case <-ctx.Done():
19        fmt.Println("Tempo esgotado!")
20    case <-ch:
21        fmt.Println("Todos os dados foram buscados!")
22    }
23}

O código acima imprime Tempo esgotado! se não conseguir obter todos os dados em 2 segundos e imprime Todos os dados foram buscados! se conseguir obter todos os dados. Ao usar context dessa forma, você pode gerenciar facilmente cancelamentos e tempos limite, mesmo em código onde várias goroutines estão operando.

Várias funções e métodos relacionados ao contexto podem ser encontrados em godoc context. Espero que você aprenda o básico e possa usá-los confortavelmente.

Channel

Channel não-bufferizado

channel é uma ferramenta para comunicação entre goroutines. channel pode ser criado com make(chan T). Aqui, T é o tipo de dados que o channel irá transferir. channel pode enviar e receber dados com <-, e pode ser fechado com close.

 1package main
 2
 3func main() {
 4    ch := make(chan int)
 5    go func() {
 6        ch <- 1
 7        ch <- 2
 8        close(ch)
 9    }()
10
11    for i := range ch {
12        fmt.Println(i)
13    }
14}

O código acima usa channel para imprimir 1 e 2. Este código simplesmente mostra o envio e recebimento de valores em um channel. No entanto, channel oferece mais recursos do que isso. Primeiro, vamos aprender sobre channel bufferizado e channel não-bufferizado. Para começar, o exemplo escrito acima é um channel não-bufferizado, onde o ato de enviar dados para o canal e o ato de receber dados devem ocorrer simultaneamente. Se essas ações não ocorrerem simultaneamente, um deadlock poderá ocorrer.

Channel bufferizado

E se o código acima não fosse uma simples saída, mas dois processos executando trabalhos pesados? Se o segundo processo travar por um longo tempo enquanto lê e executa o processo, o primeiro processo também irá parar durante esse tempo. Podemos usar channel bufferizado para evitar essa situação.

 1package main
 2
 3func main() {
 4    ch := make(chan int, 2)
 5    go func() {
 6        ch <- 1
 7        ch <- 2
 8        close(ch)
 9    }()
10
11    for i := range ch {
12        fmt.Println(i)
13    }
14}

O código acima usa channel bufferizado para imprimir 1 e 2. Neste código, usamos channel bufferizado para garantir que o ato de enviar dados para o channel e o ato de receber dados não precisem ocorrer simultaneamente. Ao adicionar um buffer ao canal, há uma folga correspondente ao seu comprimento, o que pode evitar atrasos de trabalho causados por operações subsequentes.

select

Ao lidar com vários canais, você pode usar a sintaxe select para implementar facilmente uma estrutura fan-in.

 1package main
 2
 3import (
 4    "fmt"
 5    "time"
 6)
 7
 8func main() {
 9    ch1 := make(chan int, 10)
10    ch2 := make(chan int, 10)
11    ch3 := make(chan int, 10)
12
13    go func() {
14        for {
15            ch1 <- 1
16            time.Sleep(1 * time.Second)
17        }
18    }()
19    go func() {
20        for {
21            ch2 <- 2
22            time.Sleep(2 * time.Second)
23        }
24    }()
25    go func() {
26        for {
27            ch3 <- 3
28            time.Sleep(3 * time.Second)
29        }
30    }()
31
32    for i := 0; i < 3; i++ {
33        select {
34        case v := <-ch1:
35            fmt.Println(v)
36        case v := <-ch2:
37            fmt.Println(v)
38        case v := <-ch3:
39            fmt.Println(v)
40        }
41    }
42}

O código acima cria três canais que enviam 1, 2 e 3 periodicamente, e usa select para receber e imprimir valores dos canais. Usando select dessa forma, você pode receber dados de vários canais simultaneamente e processá-los conforme os valores são recebidos dos canais.

for range

channel pode receber dados facilmente usando for range. Quando for range é usado em um canal, ele opera cada vez que dados são adicionados ao canal e encerra o loop quando o canal é fechado.

 1package main
 2
 3func main() {
 4    ch := make(chan int)
 5    go func() {
 6        ch <- 1
 7        ch <- 2
 8        close(ch)
 9    }()
10
11    for i := range ch {
12        fmt.Println(i)
13    }
14}

O código acima usa channel para imprimir 1 e 2. Neste código, usamos for range para receber e imprimir dados cada vez que dados são adicionados ao canal. E quando o canal é fechado, o loop termina.

Como escrito algumas vezes acima, essa sintaxe também pode ser usada para fins simples de sincronização.

 1package main
 2
 3func main() {
 4    ch := make(chan struct{})
 5    go func() {
 6        defer close(ch)
 7        time.Sleep(1 * time.Second)
 8        fmt.Println("Olá, Mundo!")
 9    }()
10
11    fmt.Println("Aguardando goroutine...")
12    for range ch {}
13}

O código acima imprime Olá, Mundo! após uma pausa de 1 segundo. Neste código, usamos channel para mudar o código síncrono para código assíncrono. Ao usar channel dessa forma, você pode facilmente alterar o código síncrono para o código assíncrono e definir o ponto join.

etc

Se você enviar ou receber dados em um canal nulo, você pode ficar preso em um loop infinito e ocorrer um deadlock.
Se você enviar dados depois de fechar um canal, ocorrerá um pânico.
Mesmo que você não feche um canal, o GC irá coletá-lo e fechar o canal.

mutex

spinlock

spinlock é um método de sincronização que tenta continuamente obter um lock girando em um loop. Em Go, você pode facilmente implementar um spinlock usando ponteiros.

 1package spinlock
 2
 3import (
 4    "runtime"
 5    "sync/atomic"
 6)
 7
 8type SpinLock struct {
 9    lock uintptr
10}
11
12func (s *SpinLock) Lock() {
13    for !atomic.CompareAndSwapUintptr(&s.lock, 0, 1) {
14        runtime.Gosched()
15    }
16}
17
18func (s *SpinLock) Unlock() {
19    atomic.StoreUintptr(&s.lock, 0)
20}
21
22func NewSpinLock() *SpinLock {
23    return &SpinLock{}
24}

O código acima é um código que implementa o pacote spinlock. Este código usa o pacote sync/atomic para implementar o SpinLock. O método Lock usa atomic.CompareAndSwapUintptr para tentar obter um lock, e o método Unlock usa atomic.StoreUintptr para liberar o lock. Como este método tenta obter um lock sem parar, ele continua usando a CPU até que o lock seja obtido e pode ficar preso em um loop infinito. Portanto, spinlock é recomendado para uso para sincronização simples ou para uso apenas por um curto período de tempo.

sync.Mutex

mutex é uma ferramenta para sincronização entre goroutines. O mutex fornecido pelo pacote sync fornece métodos como Lock, Unlock, RLock e RUnlock. mutex pode ser criado com sync.Mutex, e você também pode usar um lock de leitura/escrita com sync.RWMutex.

 1package main
 2
 3import (
 4    "sync"
 5)
 6
 7func main() {
 8    var mu sync.Mutex
 9    var count int
10
11    go func() {
12        mu.Lock()
13        count++
14        mu.Unlock()
15    }()
16
17    mu.Lock()
18    count++
19    mu.Unlock()
20
21    println(count)
22}

No código acima, duas goroutines acessam a mesma variável count quase ao mesmo tempo. Neste momento, se você usar um mutex para transformar o código que acessa a variável count em uma seção crítica, você pode impedir o acesso simultâneo à variável count. Então, este código sempre imprimirá 2, não importa quantas vezes seja executado.

sync.RWMutex

sync.RWMutex é um mutex que pode usar locks de leitura e escrita separadamente. Você pode usar os métodos RLock e RUnlock para aplicar e liberar um lock de leitura.

 1package cmap
 2
 3import (
 4    "sync"
 5)
 6
 7type ConcurrentMap[K comparable, V any] struct {
 8    sync.RWMutex
 9    data map[K]V
10}
11
12func (m *ConcurrentMap[K, V]) Get(key K) (V, bool) {
13    m.RLock()
14    defer m.RUnlock()
15
16    value, ok := m.data[key]
17    return value, ok
18}
19
20func (m *ConcurrentMap[K, V]) Set(key K, value V) {
21    m.Lock()
22    defer m.Unlock()
23
24    m.data[key] = value
25}

O código acima é um código que implementa ConcurrentMap usando sync.RWMutex. Neste código, um lock de leitura é aplicado no método Get e um lock de escrita é aplicado no método Set para acessar e modificar o mapa data com segurança. O motivo pelo qual um lock de leitura é necessário é que, em muitos casos de operações de leitura simples, um lock de escrita não é aplicado e apenas um lock de leitura é aplicado, permitindo que várias goroutines executem operações de leitura simultaneamente. Ao fazer isso, o desempenho pode ser melhorado aplicando apenas um lock de leitura quando nenhuma alteração de estado é necessária, eliminando a necessidade de aplicar um lock de escrita.

fakelock

fakelock é um truque simples que implementa sync.Locker. Essa estrutura fornece os mesmos métodos que sync.Mutex, mas não executa nenhuma operação real.

1package fakelock
2
3type FakeLock struct{}
4
5func (f *FakeLock) Lock() {}
6
7func (f *FakeLock) Unlock() {}

O código acima é um código que implementa o pacote fakelock. Este pacote implementa sync.Locker para fornecer os métodos Lock e Unlock, mas na verdade não faz nada. Explicarei por que este código é necessário quando tiver a oportunidade.

waitgroup

sync.WaitGroup

sync.WaitGroup é uma ferramenta para esperar até que todas as operações de goroutines sejam concluídas. Ele fornece os métodos Add, Done e Wait. O método Add adiciona o número de goroutines, o método Done notifica que uma goroutine terminou sua operação e o método Wait espera até que todas as operações de goroutines sejam concluídas.

 1package main
 2
 3import (
 4    "sync"
 5    "sync/atomic"
 6)
 7
 8func main() {
 9    wg := sync.WaitGroup{}
10    c := atomic.Int64{}
11
12    for i := 0; i < 100 ; i++ {
13        wg.Add(1)
14        go func() {
15            defer wg.Done()
16            c.Add(1)
17        }()
18    }
19
20    wg.Wait()
21    println(c.Load())
22}

O código acima usa sync.WaitGroup para fazer com que 100 goroutines adicionem um valor à variável c simultaneamente. Neste código, sync.WaitGroup é usado para esperar até que todas as goroutines terminem e, em seguida, o valor adicionado à variável c é impresso. Embora usar apenas canais seja suficiente para tarefas simples de fork & join, usar sync.WaitGroup também é uma boa opção ao lidar com uma grande quantidade de tarefas de fork & join.

com slice

Se usado com um slice, waitgroup pode ser uma excelente ferramenta para gerenciar operações simultâneas sem locks.

 1package main
 2
 3import (
 4	"fmt"
 5	"sync"
 6    "rand"
 7)
 8
 9func main() {
10	var wg sync.WaitGroup
11	arr := [10]int{}
12
13	for i := 0; i < 10; i++ {
14		wg.Add(1)
15		go func(id int) {
16			defer wg.Done()
17
18			arr[id] = rand.Intn(100)
19		}(i)
20	}
21
22	wg.Wait()
23	fmt.Println("Done")
24
25    for i, v := range arr {
26        fmt.Printf("arr[%d] = %d\n", i, v)
27    }
28}

O código acima usa apenas waitgroup para fazer com que cada goroutine gere 10 inteiros aleatórios simultaneamente e os armazene no índice atribuído. Neste código, waitgroup é usado para esperar até que todas as goroutines terminem e, em seguida, Done é impresso. Ao usar waitgroup dessa maneira, várias goroutines podem executar tarefas simultaneamente, armazenar dados sem locks até que todas as goroutines terminem e, em seguida, executar o pós-processamento de uma vez.

golang.org/x/sync/errgroup.ErrGroup

errgroup é um pacote que estende sync.WaitGroup. Ao contrário de sync.WaitGroup, se ocorrer um erro em uma das goroutines, errgroup cancela todas as goroutines e retorna o erro.

 1package main
 2
 3import (
 4    "context"
 5    "fmt"
 6    "golang.org/x/sync/errgroup"
 7)
 8
 9func main() {
10    g, ctx := errgroup.WithContext(context.Background())
11    _ = ctx
12
13    for i := 0; i < 10; i++ {
14        i := i
15        g.Go(func() error {
16            if i == 5 {
17                return fmt.Errorf("error")
18            }
19            return nil
20        })
21    }
22
23    if err := g.Wait(); err != nil {
24        fmt.Println(err)
25    }
26}

O código acima usa errgroup para criar 10 goroutines e gera um erro na quinta goroutine. O erro foi gerado intencionalmente na quinta goroutine para mostrar o caso em que ocorre um erro. No entanto, ao usá-lo na prática, você deve usar errgroup para criar goroutines e executar vários pós-processamentos para o caso em que ocorrem erros em cada goroutine.

once

É uma ferramenta para executar códigos que devem ser executados apenas uma vez. O código relacionado pode ser executado por meio dos seguintes construtores.

1func OnceFunc(f func()) func()
2func OnceValue[T any](f func() T) func() T
3func OnceValues[T1, T2 any](f func() (T1, T2)) func() (T1, T2)

OnceFunc

OnceFunc simplesmente garante que a função correspondente só possa ser executada uma vez em toda a aplicação.

 1package main
 2
 3import "sync"
 4
 5func main() {
 6    once := sync.OnceFunc(func() {
 7        println("Hello, World!")
 8    })
 9
10    once()
11    once()
12    once()
13    once()
14    once()
15}

O código acima usa sync.OnceFunc para imprimir Hello, World!. Neste código, a função once é criada usando sync.OnceFunc, e mesmo se a função once for chamada várias vezes, Hello, World! é impresso apenas uma vez.

OnceValue

OnceValue não apenas garante que a função correspondente seja executada apenas uma vez em toda a aplicação, mas também armazena o valor de retorno da função e retorna o valor armazenado quando chamada novamente.

 1package main
 2
 3import "sync"
 4
 5func main() {
 6    c := 0
 7    once := sync.OnceValue(func() int {
 8        c += 1
 9        return c
10    })
11
12    println(once())
13    println(once())
14    println(once())
15    println(once())
16    println(once())
17}

O código acima usa sync.OnceValue para incrementar a variável c em 1. Neste código, a função once é criada usando sync.OnceValue, e mesmo se a função once for chamada várias vezes, ela retorna 1, que é o valor de c incrementado apenas uma vez.

OnceValues

OnceValues funciona da mesma forma que OnceValue, mas pode retornar vários valores.

 1package main
 2
 3import "sync"
 4
 5func main() {
 6    c := 0
 7    once := sync.OnceValues(func() (int, int) {
 8        c += 1
 9        return c, c
10    })
11
12    a, b := once()
13    println(a, b)
14    a, b = once()
15    println(a, b)
16    a, b = once()
17    println(a, b)
18    a, b = once()
19    println(a, b)
20    a, b = once()
21    println(a, b)
22}

O código acima usa sync.OnceValues para incrementar a variável c em 1. Neste código, a função once é criada usando sync.OnceValues, e mesmo se a função once for chamada várias vezes, ela retorna 1, que é o valor de c incrementado apenas uma vez.

atomic

O pacote atomic é um pacote que fornece operações atômicas. O pacote atomic fornece métodos como Add, CompareAndSwap, Load, Store, Swap, etc., mas recentemente o uso de tipos como Int64, Uint64, Pointer é recomendado.

 1package main
 2
 3import (
 4    "sync"
 5    "sync/atomic"
 6)
 7
 8func main() {
 9    wg := sync.WaitGroup{}
10    c := atomic.Int64{}
11
12    for i := 0; i < 100 ; i++ {
13        wg.Add(1)
14        go func() {
15            defer wg.Done()
16            c.Add(1)
17        }()
18    }
19
20    wg.Wait()
21    println(c.Load())
22}

Este é o exemplo que foi usado anteriormente. É um código que incrementa atomicamente a variável c usando o tipo atomic.Int64. A variável pode ser incrementada e lida atomicamente com os métodos Add e Load. Além disso, os valores podem ser armazenados com o método Store, os valores podem ser trocados com o método Swap, e os valores podem ser comparados e trocados se corresponderem com o método CompareAndSwap.

cond

sync.Cond

O pacote cond é um pacote que fornece variáveis de condição. O pacote cond pode ser criado com sync.Cond e fornece os métodos Wait, Signal e Broadcast.

 1package main
 2
 3import (
 4    "sync"
 5)
 6
 7func main() {
 8    c := sync.NewCond(&sync.Mutex{})
 9    ready := false
10
11    go func() {
12        c.L.Lock()
13        ready = true
14        c.Signal()
15        c.L.Unlock()
16    }()
17
18    c.L.Lock()
19    for !ready {
20        c.Wait()
21    }
22    c.L.Unlock()
23
24    println("Ready!")
25}

O código acima usa sync.Cond para esperar até que a variável ready se torne true. Neste código, sync.Cond é usado para esperar até que a variável ready se torne true e, em seguida, Ready! é impresso. Ao usar sync.Cond dessa maneira, várias goroutines podem ser feitas para esperar simultaneamente até que uma determinada condição seja atendida.

Um queue simples pode ser implementado usando isso.

 1package queue
 2
 3import (
 4    "sync"
 5    "sync/atomic"
 6)
 7
 8type Node[T any] struct {
 9    Value T
10    Next  *Node[T]
11}
12
13type Queue[T any] struct {
14    sync.Mutex
15    Cond *sync.Cond
16    Head *Node[T]
17    Tail *Node[T]
18    Len  int
19}
20
21func New[T any]() *Queue[T] {
22    q := &Queue[T]{}
23    q.Cond = sync.NewCond(&q.Mutex)
24    return q
25}
26
27func (q *Queue[T]) Push(value T) {
28    q.Lock()
29    defer q.Unlock()
30
31    node := &Node[T]{Value: value}
32    if q.Len == 0 {
33        q.Head = node
34        q.Tail = node
35    } else {
36        q.Tail.Next = node
37        q.Tail = node
38    }
39    q.Len++
40    q.Cond.Signal()
41}
42
43func (q *Queue[T]) Pop() T {
44    q.Lock()
45    defer q.Unlock()
46
47    for q.Len == 0 {
48        q.Cond.Wait()
49    }
50
51    node := q.Head
52    q.Head = q.Head.Next
53    q.Len--
54    return node.Value
55}

Ao usar sync.Cond dessa forma, você pode esperar com eficiência e retomar a operação quando a condição for atendida, em vez de usar muito uso da CPU com spin-lock.

semaphore

golang.org/x/sync/semaphore.Semaphore

O pacote semaphore é um pacote que fornece semáforos. O pacote semaphore pode ser criado com golang.org/x/sync/semaphore.Semaphore e fornece os métodos Acquire, Release e TryAcquire.

 1package main
 2
 3import (
 4    "context"
 5    "fmt"
 6    "golang.org/x/sync/semaphore"
 7)
 8
 9func main() {
10    s := semaphore.NewWeighted(1)
11
12    if s.TryAcquire(1) {
13        fmt.Println("Acquired!")
14    } else {
15        fmt.Println("Not Acquired!")
16    }
17
18    s.Release(1)
19}

O código acima usa semaphore para criar um semáforo, usa o semáforo para adquirir o semáforo com o método Acquire e libera o semáforo com o método Release. Este código mostrou como adquirir e liberar um semáforo usando semaphore.

Em conclusão

Acho que o básico é o suficiente até aqui. Com base no conteúdo deste artigo, espero que você entenda como gerenciar a simultaneidade usando goroutines e seja capaz de usá-lo na prática. Espero que este artigo tenha sido útil para você. Obrigado.