Go Concurrency Starter Pack
Przegląd
Krótkie wprowadzenie
Język Go oferuje wiele narzędzi do zarządzania współbieżnością. W tym artykule przedstawimy niektóre z nich oraz związane z nimi techniki.
Goroutine?
Goroutine to nowy rodzaj modelu współbieżności obsługiwany przez język Go. Zazwyczaj program otrzymuje wątki OS od systemu operacyjnego do wykonywania wielu zadań jednocześnie i wykonuje zadania równolegle w liczbie odpowiadającej liczbie rdzeni. Aby wykonywać współbieżność w mniejszych jednostkach, w przestrzeni użytkownika (userland) tworzone są zielone wątki (green threads), co pozwala wielu zielonym wątkom na wykonywanie zadań w ramach jednego wątku OS. Jednak w przypadku goroutines, te zielone wątki zostały stworzone w jeszcze mniejszej i bardziej efektywnej formie. Goroutines zużywają mniej pamięci niż wątki i mogą być tworzone i przełączane szybciej niż wątki.
Aby użyć goroutine, wystarczy użyć słowa kluczowego go. Pozwala to na intuicyjne wykonywanie kodu synchronicznego jako kodu asynchronicznego podczas pisania programu.
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 wyświetla Hello, World!, na przepływ asynchroniczny. Chociaż obecny przykład jest prosty, przekształcenie nieco bardziej złożonego kodu synchronicznego na asynchroniczny poprawia czytelność, widoczność i zrozumiałość kodu w porównaniu z istniejącymi metodami takimi jak async/await czy Promise.
Należy jednak pamiętać, że w wielu przypadkach, jeśli nie rozumie się przepływu prostego wywoływania kodu synchronicznego asynchronicznie oraz przepływu fork & join (podobnego do dziel i zwyciężaj), może powstać słaby kod goroutine. W tym artykule przedstawimy kilka metod i technik, które mogą pomóc w uniknięciu takich sytuacji.
Zarządzanie współbieżnością
context
Pojawienie się context jako pierwszej techniki zarządzania może być zaskakujące. Jednak w języku Go context odgrywa doskonałą rolę nie tylko w prostym anulowaniu, ale także w zarządzaniu całym drzewem zadań. Dla tych, którzy nie są zaznajomieni, krótko wyjaśnię 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 używa context do wyświetlenia Context is done! po 1 sekundzie. context umożliwia sprawdzenie statusu anulowania za pomocą metody Done() i oferuje różne metody anulowania, takie jak WithCancel, WithTimeout, WithDeadline i WithValue.
Stwórzmy prosty przykład. Załóżmy, że piszesz kod, który pobiera dane user, post i comment przy użyciu wzorca aggregator. Jeśli wszystkie żądania muszą zostać zrealizowane w ciągu 2 sekund, możesz to zapisać w następujący sposób:
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 wyświetla Timeout!, jeśli wszystkie dane nie zostaną pobrane w ciągu 2 sekund, a All data is fetched!, jeśli wszystkie dane zostaną pobrane. W ten sposób, używając context, można łatwo zarządzać anulowaniem i limitami czasu nawet w kodzie, w którym działa wiele goroutines.
Różne powiązane funkcje i metody context są dostępne na stronie godoc context. Mam nadzieję, że nauczysz się tych prostych i będziesz mógł z nich swobodnie korzystać.
channel
unbuffered channel
channel to narzędzie do komunikacji między goroutines. channel można utworzyć za pomocą make(chan T). W tym przypadku T to typ danych, które channel będzie przesyłać. channel może wysyłać i odbierać dane za pomocą <-, a channel można zamknąć 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 używa channel do wyświetlania 1 i 2. Ten kod pokazuje tylko wysyłanie i odbieranie wartości w channel. Jednak channel oferuje więcej funkcji. Najpierw omówimy buffered channel i unbuffered channel. Zanim zaczniemy, przykład powyżej jest unbuffered channel, co oznacza, że wysyłanie danych do kanału i odbieranie danych z kanału musi odbywać się jednocześnie. Jeśli te akcje nie będą wykonywane jednocześnie, może wystąpić zakleszczenie (deadlock).
buffered channel
A co jeśli powyższy kod nie jest prostym wyświetlaniem, ale dwoma procesami wykonującymi ciężkie zadania? Jeśli drugi proces zawiesi się na dłuższy czas podczas odczytu i przetwarzania, pierwszy proces również zostanie zatrzymany na ten czas. Aby zapobiec takiej sytuacji, możemy 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 używa buffered channel do wyświetlania 1 i 2. W tym kodzie, używając buffered channel, wysyłanie danych do channel i odbieranie danych z channel nie muszą odbywać się jednocześnie. Dzięki buforowi w kanale, uzyskuje się pewną swobodę, która zapobiega opóźnieniom zadań spowodowanym przez wpływ zadań o niższym priorytecie.
select
Podczas pracy 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 przesyłają 1, 2 i 3, a następnie używa select do odbierania i wyświetlania wartości z kanałów. W ten sposób, używając select, można jednocześnie odbierać dane z wielu kanałów i przetwarzać je, gdy tylko wartości zostaną odebrane z kanału.
for range
channel może łatwo odbierać dane za pomocą for range. Użycie for range z kanałem spowoduje jego uruchomienie za każdym razem, gdy dane zostaną dodane do kanału, a pętla zakończy się po zamknięciu kanału.
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 używa channel do wyświetlania 1 i 2. W tym kodzie, używając for range, dane są odbierane i wyświetlane za każdym razem, gdy dane zostaną dodane do kanału. Po zamknięciu kanału pętla się kończy.
Jak wspomniano kilkakrotnie powyżej, ta składnia może być również używana jako proste narzędzie 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 wyświetla "Hello, World!" po sekundzie. W tym kodzie channel został użyty do zmiany kodu synchronicznego na kod asynchroniczny. W ten sposób, używając channel, można łatwo zmienić kod synchroniczny na kod asynchroniczny i ustawić punkty join.
etc
- Wysyłanie lub odbieranie danych z kanału
nilmoże prowadzić do nieskończonej pętli i zakleszczenia. - Wysyłanie danych po zamknięciu kanału spowoduje
panic. - Kanał nie musi być zamykany ręcznie, ponieważ Garbage Collector (GC) zamknie go podczas zbierania.
mutex
spinlock
spinlock to metoda synchronizacji, która polega na ciągłym próbowaniu blokady w pętli. W języku Go można łatwo zaimplementować spinlock za pomocą 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 implementuje pakiet spinlock. W tym kodzie SpinLock jest zaimplementowany przy użyciu pakietu sync/atomic. Metoda Lock próbuje zablokować za pomocą atomic.CompareAndSwapUintptr, a metoda Unlock zwalnia blokadę za pomocą atomic.StoreUintptr. Ta metoda ciągle próbuje uzyskać blokadę bez przerwy, co powoduje ciągłe użycie procesora, dopóki blokada nie zostanie uzyskana, co może prowadzić do nieskończonej pętli. Dlatego spinlock jest zalecany do prostych synchronizacji lub do użycia przez krótki czas.
sync.Mutex
mutex to narzędzie do synchronizacji między goroutines. mutex dostarczany przez pakiet sync oferuje metody takie jak Lock, Unlock, RLock, RUnlock. mutex można utworzyć za pomocą sync.Mutex, a do blokad 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 goroutines prawie jednocześnie uzyskują dostęp do tej samej zmiennej count. W tym przypadku, używając mutex do utworzenia sekcji krytycznej dla kodu, który uzyskuje dostęp do zmiennej count, można zapobiec jednoczesnemu dostępowi do zmiennej count. Wtedy ten kod zawsze wyświetli 2, niezależnie od tego, ile razy zostanie uruchomiony.
sync.RWMutex
sync.RWMutex to mutex, który pozwala na rozróżnienie między blokadami odczytu a blokadami zapisu. Metody RLock i RUnlock mogą być użyte do zablokowania i zwolnienia blokady 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 przy użyciu sync.RWMutex. W tym kodzie metoda Get blokuje odczyt, a metoda Set blokuje zapis, co pozwala na bezpieczny dostęp i modyfikację mapy data. Powodem, dla którego potrzebna jest blokada odczytu, jest to, że w przypadku wielu prostych operacji odczytu, wiele goroutines może jednocześnie wykonywać operacje odczytu, blokując tylko odczyt, bez blokady zapisu. Dzięki temu, w przypadkach, gdy nie ma potrzeby blokady zapisu, ponieważ nie ma zmian stanu, można zastosować tylko blokadę odczytu, aby poprawić wydajność.
fakelock
fakelock to prosta sztuczka implementująca sync.Locker. Ta struktura zapewnia te same metody co sync.Mutex, ale nie wykonuje żadnych rzeczywistych operacji.
1package fakelock
2
3type FakeLock struct{}
4
5func (f *FakeLock) Lock() {}
6
7func (f *FakeLock) Unlock() {}
Powyższy kod implementuje pakiet fakelock. Ten pakiet implementuje sync.Locker, zapewniając metody Lock i Unlock, ale w rzeczywistości nie wykonuje żadnych operacji. Dlaczego taki kod jest potrzebny, wyjaśnię, jeśli nadarzy się okazja.
waitgroup
sync.WaitGroup
sync.WaitGroup to narzędzie, które czeka, aż wszystkie goroutines zakończą swoje zadania. Oferuje metody Add, Done i Wait. Metoda Add dodaje liczbę goroutines, a metoda Done informuje, że goroutine zakończyła swoje zadanie. Metoda Wait czeka, aż wszystkie goroutines zakończą swoje zadania.
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 używa sync.WaitGroup do dodawania wartości do zmiennej c jednocześnie przez 100 goroutines. W tym kodzie, używając sync.WaitGroup, czeka się, aż wszystkie goroutines zakończą, a następnie wyświetla się wartość dodaną do zmiennej c. Chociaż do łączenia kilku zadań wystarczyłby tylko kanał, w przypadku łączenia dużej liczby zadań sync.WaitGroup jest również dobrym wyborem.
with slice
W połączeniu ze slice, waitgroup może być doskonałym narzędziem do zarządzania równoległymi operacjami 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 używa tylko waitgroup do jednoczesnego generowania 10 losowych liczb całkowitych przez każdą goroutine i zapisywania ich w przypisanym indeksie. W tym kodzie, używając waitgroup, czeka się, aż wszystkie goroutines zakończą, a następnie wyświetla się „Done”. W ten sposób, używając waitgroup, wiele goroutines może jednocześnie wykonywać zadania, zapisywać dane bez blokad, aż wszystkie goroutines zakończą, a następnie wykonywać przetwarzanie końcowe wsadowo po zakończeniu zadań.
golang.org/x/sync/errgroup.ErrGroup
errgroup to rozszerzony pakiet sync.WaitGroup. W przeciwieństwie do sync.WaitGroup, errgroup anuluje wszystkie goroutines i zwraca błąd, jeśli wystąpi błąd w którejkolwiek z goroutines.
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 używa errgroup do utworzenia 10 goroutines i celowo generuje błąd w 5. goroutine, aby pokazać przypadek wystąpienia błędu. Jednak w rzeczywistym użyciu, errgroup jest używany do tworzenia goroutines, a w przypadku wystąpienia błędu w każdej goroutine, można przeprowadzić różne przetwarzania końcowe.
once
Narzędzie do wykonania kodu 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 zapewnia, że dana funkcja zostanie wykonana tylko 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 używa sync.OnceFunc do wyświetlania Hello, World!. W tym kodzie, używając sync.OnceFunc, tworzona jest funkcja once, a nawet jeśli funkcja once zostanie wywołana wiele razy, Hello, World! zostanie wyświetlone tylko raz.
OnceValue
OnceValue zapewnia, że dana funkcja zostanie wykonana tylko raz w całym programie, a także przechowuje wartość zwracaną przez tę funkcję i zwraca ją przy ponownym wywołaniu.
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 używa sync.OnceValue do zwiększania zmiennej c o 1. W tym kodzie, używając sync.OnceValue, tworzona jest funkcja once, a nawet jeśli funkcja once zostanie wywołana wiele razy, zmienna c zostanie zwiększona tylko raz i zwróci 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 używa sync.OnceValues do zwiększania zmiennej c o 1. W tym kodzie, używając sync.OnceValues, tworzona jest funkcja once, a nawet jeśli funkcja once zostanie wywołana wiele razy, zmienna c zostanie zwiększona tylko raz i zwróci 1.
atomic
Pakiet atomic zapewnia operacje atomowe. Pakiet atomic oferuje metody takie jak Add, CompareAndSwap, Load, Store, Swap, ale ostatnio zaleca się używanie 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 jest przykład, który był użyty wcześniej. Jest to kod, który atomowo zwiększa zmienną c za pomocą typu atomic.Int64. Metody Add i Load mogą być używane do atomowego zwiększania zmiennej i odczytywania jej wartości. Ponadto, metoda Store może być używana do zapisywania wartości, metoda Swap do wymiany wartości, a metoda CompareAndSwap do porównywania i wymiany wartości, jeśli jest to odpowiednie.
cond
sync.Cond
Pakiet cond zapewnia zmienne warunkowe. Pakiet cond można utworzyć za pomocą sync.Cond i oferuje metody Wait, Signal, 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 używa sync.Cond do oczekiwania, aż zmienna ready stanie się true. W tym kodzie, używając sync.Cond, oczekuje się, aż zmienna ready stanie się true, a następnie wyświetla się Ready!. W ten sposób, używając sync.Cond, można spowodować, że wiele goroutines będzie czekać, aż określony warunek zostanie spełniony.
Można to wykorzystać do zaimplementowania prostej 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}
W ten sposób, wykorzystując sync.Cond, można efektywnie oczekiwać i wznawiać działanie, gdy warunek zostanie spełniony, zamiast zużywać dużo mocy procesora przez spin-lock.
semaphore
golang.org/x/sync/semaphore.Semaphore
Pakiet semaphore dostarcza semafor. Pakiet semaphore można utworzyć za pomocą golang.org/x/sync/semaphore.Semaphore i oferuje metody Acquire, Release, 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 używa semaphore do utworzenia semafora, a następnie używa metod Acquire do jego uzyskania i Release do jego zwolnienia. W tym kodzie pokazaliśmy, jak uzyskać i zwolnić semafor za pomocą semaphore.
Podsumowanie
Myślę, że podstawowe informacje wystarczą. Mam nadzieję, że na podstawie treści tego artykułu zrozumiecie, jak zarządzać współbieżnością za pomocą goroutines, i będziecie mogli z nich korzystać w praktyce. Mam nadzieję, że ten artykuł był dla Was pomocny. Dziękuję.