GoSudaPakiet startowy do współbieżności w Go
Przegląd
Krótkie wprowadzenie
Język Go posiada wiele narzędzi do zarządzania współbieżnością. W tym artykule przedstawimy niektóre z nich oraz kilka sztuczek.
Goroutine?
goroutine to nowy rodzaj modelu współbieżności obsługiwanego przez język Go. Zwykle program otrzymuje od systemu operacyjnego wątki systemu operacyjnego (OS threads), aby równolegle wykonywać wiele zadań, w liczbie odpowiadającej liczbie rdzeni. Aby realizować współbieżność na mniejszą skalę, w przestrzeni użytkownika (userland) tworzy się zielone wątki (green threads), dzięki czemu wiele zielonych wątków wykonuje zadania naprzemiennie w obrębie jednego wątku systemu operacyjnego. Goroutine sprawiają jednak, że ta forma zielonych wątków jest jeszcze mniejsza i bardziej wydajna. Goroutine zużywają mniej pamięci niż wątki i mogą być tworzone oraz przełączane szybciej niż wątki.
Aby użyć goroutine, wystarczy zastosować słowo kluczowe go. Pozwala to w sposób intuicyjny podczas pisania programu wykonywać kod synchroniczny jako kod asynchroniczny.
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("Hello, World!")
14 }()
15
16 fmt.Println("Waiting for goroutine...")
17 for range ch {}
18}
Ten kod w prosty sposób zmienia synchroniczny kod, który czeka 1 sekundę, a następnie wypisuje Hello, World!, na przepływ asynchroniczny. Obecny przykład jest prosty, ale jeśli nieco bardziej złożony kod zostanie przekształcony z synchronicznego na asynchroniczny, czytelność, przejrzystość i zrozumiałość kodu będą lepsze niż w dotychczasowych podejściach, takich jak async await czy promise.
Jednak w wielu przypadkach, bez zrozumienia przepływu polegającego na prostym wywoływaniu takiego kodu synchronicznego w sposób asynchroniczny oraz przepływu przypominającego fork & join (podobnego do dziel i zwyciężaj), powstaje słaby kod z goroutine. W tym artykule przedstawimy kilka metod i technik, które pomagają przygotować się na takie przypadki.
Zarządzanie współbieżnością
context
To, że pierwszą technikę zarządzania stanowi context, może być zaskoczeniem. Jednak w języku Go context wykracza poza prostą funkcję anulowania i doskonale sprawdza się w zarządzaniu całym drzewem zadań. Dla osób, które go nie znają, krótko opiszę ten pakiet.
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("Context is done!")
10 }()
11
12 time.Sleep(1 * time.Second)
13
14 cancel()
15
16 time.Sleep(1 * time.Second)
17}
Powyższy kod za pomocą context wypisuje Context is done! po 1 sekundzie. context pozwala sprawdzić, czy nastąpiło anulowanie, za pomocą metody Done(), a metody takie jak WithCancel, WithTimeout, WithDeadline i WithValue zapewniają różne sposoby anulowania.
Stwórzmy prosty przykład. Załóżmy, że chcesz napisać kod pobierający dane user, post i comment z wykorzystaniem wzorca aggregator. Jeśli wszystkie żądania muszą zakończyć się w ciągu 2 sekund, możesz napisać to następująco.
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("Timeout!")
20 case <-ch:
21 fmt.Println("All data is fetched!")
22 }
23}
Powyższy kod wypisuje Timeout!, jeśli nie uda się pobrać wszystkich danych w ciągu 2 sekund, a jeśli wszystkie dane zostaną pobrane, wypisuje All data is fetched!. Używając context w ten sposób, możesz łatwo zarządzać anulowaniem i limitami czasu w kodzie, w którym działa wiele goroutine.
Różne funkcje i metody związane z context są dostępne w godoc context. Mam nadzieję, że po przyswojeniu prostszych elementów można będzie z nich wygodnie korzystać.
channel
unbuffered channel
channel to narzędzie do komunikacji między goroutine. channel można utworzyć za pomocą make(chan T). Tutaj T oznacza typ danych, które dany channel będzie przesyłał. Dane wysyła się i odbiera za pomocą <-, a channel zamyka się za pomocą 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}
Powyższy kod za pomocą channel wypisuje 1 i 2. W tym kodzie pokazano jedynie wysyłanie i odbieranie wartości przez channel. Jednak channel oferuje więcej możliwości. Najpierw przyjrzyjmy się buffered channel i unbuffered channel. Zanim zaczniemy, zauważmy, że przedstawiony przykład używa unbuffered channel, w którym wysłanie danych do kanału i ich odebranie muszą nastąpić jednocześnie. Jeśli te działania nie zajdą w tym samym czasie, może dojść do deadlocka.
buffered channel
A co, jeśli powyższy kod wykonuje nie proste wypisywanie, lecz dwa procesy wykonujące ciężkie zadania? Jeśli drugi proces odczytuje i przetwarza dane, a to trwa długo, pierwszy proces również zostanie zatrzymany na ten czas. Aby zapobiec takiej sytuacji, można użyć buffered channel.
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}
Powyższy kod za pomocą buffered channel wypisuje 1 i 2. W tym kodzie buffered channel sprawia, że wysłanie danych do channel i ich odebranie nie muszą nastąpić jednocześnie. Dzięki buforowi w kanale powstaje zapas o takiej długości, co pozwala zapobiegać opóźnieniom pracy wynikającym z wpływu zadań o niższym priorytecie.
select
Pracując z wieloma kanałami, składnia select pozwala łatwo zaimplementować strukturę 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}
Powyższy kod tworzy 3 kanały, które cyklicznie przekazują wartości 1, 2 i 3, a następnie za pomocą select odbiera wartości z kanałów i je wypisuje. Dzięki select można jednocześnie odbierać dane z wielu kanałów i przetwarzać je w miarę ich nadejścia.
for range
channel można łatwo odczytywać za pomocą for range. Gdy for range zostanie użyte z kanałem, wykonuje się za każdym razem, gdy do kanału zostanie dodana wartość, a po zamknięciu kanału pętla się kończy.
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}
Powyższy kod za pomocą channel wypisuje 1 i 2. W tym kodzie for range odbiera i wypisuje dane za każdym razem, gdy zostaną dodane do kanału. Gdy kanał zostanie zamknięty, pętla się kończy.
Jak wspomniano kilka razy wyżej, tę składnię można także użyć jako prostego mechanizmu synchronizacji.
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("Hello, World!")
9 }()
10
11 fmt.Println("Waiting for goroutine...")
12 for range ch {}
13}
Powyższy kod czeka 1 sekundę, a następnie wypisuje Hello, World!. W tym kodzie channel został użyty do przekształcenia kodu synchronicznego w asynchroniczny. W ten sposób za pomocą channel można łatwo przekształcić kod synchroniczny w asynchroniczny oraz ustawić punkt join.
etc
- Wysłanie lub odebranie danych z nil channel może spowodować nieskończoną pętlę i deadlock.
- Wysłanie danych do zamkniętego kanału powoduje panic.
- Nawet jeśli kanału nie zamkniemy jawnie, GC go usunie, zamykając go przy okazji.
mutex
spinlock
spinlock to metoda synchronizacji, która w pętli wielokrotnie próbuje zdobyć blokadę. W języku Go można łatwo zaimplementować spinlock z użyciem wskaźników.
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}
Powyższy kod to implementacja pakietu spinlock. Wykorzystano w nim pakiet sync/atomic do zaimplementowania SpinLock. Metoda Lock próbuje zdobyć blokadę za pomocą atomic.CompareAndSwapUintptr, a metoda Unlock zwalnia blokadę za pomocą atomic.StoreUintptr. Ponieważ ta metoda bez przerwy próbuje zdobyć blokadę, do czasu jej uzyskania stale zużywa procesor i może wpaść w nieskończoną pętlę. Dlatego spinlock najlepiej stosować do prostej synchronizacji lub tylko przez krótki czas.
sync.Mutex
mutex to narzędzie do synchronizacji między goroutine. mutex dostarczany przez pakiet sync oferuje metody Lock, Unlock, RLock, RUnlock i inne. mutex można utworzyć jako sync.Mutex, a do blokady odczytu/zapisu można użyć 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}
W powyższym kodzie dwie goroutine niemal jednocześnie uzyskują dostęp do tej samej zmiennej count. Jeśli kod uzyskujący dostęp do zmiennej count ujmiemy w sekcję krytyczną za pomocą mutex, można zapobiec równoczesnemu dostępowi do count. Wtedy ten kod za każdym razem wypisze 2, niezależnie od liczby uruchomień.
sync.RWMutex
sync.RWMutex to mutex, który pozwala rozróżniać blokadę odczytu i blokadę zapisu. Za pomocą metod RLock i RUnlock można założyć i zwolnić blokadę odczytu.
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}
Powyższy kod implementuje ConcurrentMap z użyciem sync.RWMutex. W tym kodzie metoda Get zakłada blokadę odczytu, a metoda Set blokadę zapisu, dzięki czemu można bezpiecznie uzyskiwać dostęp do mapy data i ją modyfikować. Blokada odczytu jest potrzebna, gdy wykonuje się wiele prostych operacji odczytu: zamiast blokować zapis, zakłada się tylko blokadę odczytu, aby wiele goroutine mogło jednocześnie czytać. Dzięki temu w przypadkach, gdy stan nie ulega zmianie i nie trzeba zakładać blokady zapisu, samo założenie blokady odczytu pozwala poprawić wydajność.
fakelock
fakelock to prosty trik polegający na implementacji sync.Locker. Ta struktura udostępnia te same metody co sync.Mutex, ale faktycznie nic nie robi.
1package fakelock
2
3type FakeLock struct{}
4
5func (f *FakeLock) Lock() {}
6
7func (f *FakeLock) Unlock() {}
Powyższy kod to implementacja pakietu fakelock. Pakiet implementuje sync.Locker i udostępnia metody Lock oraz Unlock, ale w rzeczywistości nie wykonują one żadnych działań. Dlaczego taki kod jest potrzebny, wyjaśnię innym razem.
waitgroup
sync.WaitGroup
sync.WaitGroup to narzędzie, które czeka, aż wszystkie zadania goroutine zostaną zakończone. Udostępnia metody Add, Done i Wait: metodą Add dodaje się liczbę goroutine, metodą Done sygnalizuje się zakończenie zadania goroutine, a metodą Wait czeka się, aż zakończą się wszystkie goroutine.
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}
Powyższy kod za pomocą sync.WaitGroup pozwala 100 goroutine jednocześnie dodawać wartości do zmiennej c. W tym kodzie sync.WaitGroup czeka, aż wszystkie goroutine zakończą pracę, a następnie wypisuje wartość dodaną do c. W przypadku prostego fork & join kilku zadań wystarczy sam kanał, ale przy fork & join dużej liczby zadań dobrym wyborem jest sync.WaitGroup.
with slice
W połączeniu z sliceem waitgroup może stać się doskonałym narzędziem do zarządzania równoległym wykonywaniem zadań bez użycia blokad.
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}
Powyższy kod za pomocą samego waitgroup pozwala każdej z goroutine jednocześnie wygenerować 10 losowych liczb całkowitych i zapisać je pod przydzielonym indeksem. W tym kodzie waitgroup czeka, aż wszystkie goroutine zakończą pracę, a następnie wypisuje Done. W ten sposób za pomocą waitgroup wiele goroutine może jednocześnie wykonywać zadania, zapisywać dane bez blokad, aż wszystkie się zakończą, i dopiero potem zbiorczo przeprowadzić dalsze przetwarzanie.
golang.org/x/sync/errgroup.ErrGroup
errgroup to pakiet rozszerzający sync.WaitGroup. W przeciwieństwie do sync.WaitGroup, errgroup w razie wystąpienia błędu w choćby jednym zadaniu goroutine anuluje wszystkie goroutine i zwraca błąd.
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}
Powyższy kod za pomocą errgroup tworzy 10 goroutine, z których piąta zgłasza błąd. Celowo wywołano błąd w piątej goroutine, aby pokazać sytuację, w której błąd występuje. W praktyce errgroup używa się do tworzenia goroutine i stosowania różnorodnych dalszych działań w przypadku wystąpienia błędu w każdej z nich.
once
Narzędzie do wykonywania kodu, który powinien zostać uruchomiony tylko raz. Powiązany kod można uruchomić za pomocą poniższych konstruktorów.
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 po prostu zapewnia, że dana funkcja zostanie wykonana dokładnie raz w całym programie.
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}
Powyższy kod za pomocą sync.OnceFunc wypisuje Hello, World!. W tym kodzie sync.OnceFunc tworzy funkcję once, a mimo wielokrotnego wywołania once tekst Hello, World! zostaje wypisany tylko raz.
OnceValue
OnceValue nie tylko zapewnia, że dana funkcja wykona się dokładnie raz w całym programie, ale także zapisuje jej wartość zwrotną i przy kolejnych wywołaniach zwraca zapisaną wartość.
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}
Powyższy kod za pomocą sync.OnceValue zwiększa zmienną c o 1. W tym kodzie sync.OnceValue tworzy funkcję once, a mimo wielokrotnego jej wywołania zmienna c zwiększa się tylko raz i zwracana jest wartość 1.
OnceValues
OnceValues działa tak samo jak OnceValue, ale może zwracać wiele wartości.
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}
Powyższy kod za pomocą sync.OnceValues zwiększa zmienną c o 1. W tym kodzie sync.OnceValues tworzy funkcję once, a mimo wielokrotnego jej wywołania zmienna c zwiększa się tylko raz i zwracana jest wartość 1.
atomic
Pakiet atomic udostępnia operacje atomowe. Oferuje metody Add, CompareAndSwap, Load, Store, Swap i inne, ale ostatnio zaleca się korzystanie z typów takich jak Int64, Uint64, Pointer.
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}
To przykład, który pojawił się wcześniej. Używa typu atomic.Int64, aby atomowo zwiększać zmienną c. Metodami Add i Load można atomowo zwiększać zmienną i ją odczytywać. Ponadto metodą Store zapisuje się wartość, metodą Swap zamienia się wartość, a metodą CompareAndSwap porównuje się wartość i, jeśli pasuje, ją podmienia.
cond
sync.Cond
Pakiet cond dostarcza zmienne warunkowe. Można go utworzyć jako sync.Cond i udostępnia metody Wait, Signal i 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}
Powyższy kod za pomocą sync.Cond czeka, aż zmienna ready przyjmie wartość true. W tym kodzie sync.Cond czeka, aż ready stanie się true, a następnie wypisuje Ready!. Dzięki sync.Cond wiele goroutine może czekać, aż zostanie spełniony określony warunek.
Wykorzystując to, można zaimplementować prostą queue.
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}
Dzięki sync.Cond zamiast zużywać dużo procesora w spin-lock, można efektywnie czekać, a po spełnieniu warunku wznowić działanie.
semaphore
golang.org/x/sync/semaphore.Semaphore
Pakiet semaphore dostarcza semafory. Można go utworzyć jako golang.org/x/sync/semaphore.Semaphore, a udostępnia metody Acquire, Release i 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}
Powyższy kod tworzy semafor za pomocą semaphore, a następnie pozyskuje go metodą Acquire i zwalnia metodą Release. W tym kodzie pokazano, jak pozyskiwać i zwalniać semafor za pomocą semaphore.
Zakończenie
Na tym etapie wystarczy podstawowa wiedza. Mam nadzieję, że na podstawie tego artykułu zrozumiesz, jak zarządzać współbieżnością za pomocą goroutine, i będziesz mógł/mogła z tego korzystać w praktyce. Mam nadzieję, że ten artykuł był dla Ciebie pomocny. Dziękuję.