Go Súbežný Štartovací Balík
Prehľad
Stručný úvod
Jazyk Go má mnoho nástrojov na správu súbežnosti. V tomto článku si predstavíme niektoré z nich a triky, ktoré s nimi súvisia.
Gorutiny?
Gorutina je nový typ modelu súbežnosti podporovaný jazykom Go. Všeobecne platí, že programy získavajú vlákna OS od operačného systému, aby vykonávali viacero úloh súčasne, pričom tieto úlohy sa vykonávajú paralelne v závislosti od počtu jadier. A na vykonávanie súbežnosti menších jednotiek sa v používateľskom priestore vytvárajú zelené vlákna, takže viacero zelených vlákien môže rotovať a vykonávať úlohy v rámci jedného vlákna OS. V prípade gorutín sú však tieto zelené vlákna ešte menšie a efektívnejšie. Tieto gorutiny spotrebúvajú menej pamäte ako vlákna a dajú sa vytvárať a prepínať rýchlejšie ako vlákna.
Ak chcete používať gorutiny, stačí použiť kľúčové slovo go. To umožňuje intuitívne spúšťať synchrónny kód ako asynchrónny kód v procese písania 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}
Tento kód jednoducho zmení synchrónny kód, ktorý po 1 sekunde vypíše Hello, World!, na asynchrónny tok. Tento príklad je jednoduchý, ale ak sa trochu zložitejší kód zmení zo synchrónneho na asynchrónny, čitateľnosť, prehľadnosť a zrozumiteľnosť kódu sú lepšie ako pri existujúcich metódach async await alebo promise.
V mnohých prípadoch sa však stáva, že bez pochopenia toku jednoduchého asynchrónneho volania tohto synchrónneho kódu a toku typu fork & join (tok podobný metóde rozdeľuj a panuj) vznikne nevhodný kód gorutín. V tomto článku predstavíme niekoľko metód a techník, ktoré vám pomôžu pripraviť sa na takéto prípady.
Správa súbežnosti
context
Možno je prekvapujúce, že prvou technikou správy je context. V jazyku Go však context nepredstavuje iba funkciu zrušenia, ale zohráva aj vynikajúcu úlohu pri správe celého stromu úloh. Ak s týmto balíkom nie ste oboznámení, stručne ho popíšeme.
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}
Vyššie uvedený kód používa context na výpis Context is done! po 1 sekunde. context dokáže overiť, či sa má zrušiť prostredníctvom metódy Done() a poskytuje rôzne spôsoby zrušenia prostredníctvom metód WithCancel, WithTimeout, WithDeadline, WithValue atď.
Vytvorme si jednoduchý príklad. Predpokladajme, že píšete kód, ktorý používa vzor aggregator na získanie údajov user, post, comment na načítanie určitých údajov. Ak všetky požiadavky musia byť splnené do 2 sekúnd, môžete to napísať takto:
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}
Vyššie uvedený kód vypíše Timeout!, ak sa všetky údaje nepodarí načítať do 2 sekúnd, a vypíše All data is fetched!, ak sa všetky údaje načítajú. Použitím context týmto spôsobom môžete jednoducho spravovať zrušenie a časový limit aj v kóde, kde beží viacero gorutín.
Rôzne funkcie a metódy súvisiace s kontextom si môžete pozrieť v godoc context. Dúfame, že sa naučíte používať jednoduché funkcie a metódy, aby ste ich mohli pohodlne používať.
channel
unbuffered channel
channel je nástroj na komunikáciu medzi gorutinami. channel sa dá vytvoriť pomocou make(chan T). V tomto prípade je T typ údajov, ktoré daný channel bude odovzdávať. channel dokáže odosielať a prijímať údaje pomocou <- a channel sa dá zatvoriť pomocou 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}
Vyššie uvedený kód používa channel na výpis čísiel 1 a 2. Tento kód jednoducho ukazuje iba odosielanie a prijímanie hodnôt na channel. channel však poskytuje oveľa viac funkcií. Najprv si povieme niečo o buffered channel a unbuffered channel. Na úvod, vyššie uvedený príklad je unbuffered channel, ktorý vyžaduje, aby odosielanie a prijímanie údajov na kanáli prebiehalo súčasne. Ak sa tieto akcie neuskutočnia súčasne, môže dôjsť k zablokovaniu.
buffered channel
Čo ak vyššie uvedený kód nepredstavuje jednoduchý výpis, ale dva procesy vykonávajúce náročnú úlohu? Ak sa druhý proces pri čítaní a spracovaní dlho zdrží, zastaví sa aj prvý proces. Na zabránenie takejto situácie môžeme použiť 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}
Vyššie uvedený kód používa buffered channel na výpis čísiel 1 a 2. V tomto kóde sme použili buffered channel, aby sme zabezpečili, že odosielanie a prijímanie údajov na channel nemusia prebiehať súčasne. Pridaním vyrovnávacej pamäte do kanála týmto spôsobom, sa vytvorí rezerva s určitou dĺžkou, ktorá zabráni oneskoreniu úlohy spôsobenému následnou úlohou.
select
Pri práci s viacerými kanálmi môžete jednoducho implementovať štruktúru fan-in pomocou príkazu select.
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}
Vyššie uvedený kód vytvorí 3 kanály, ktoré pravidelne odosielajú 1, 2 a 3, a používa príkaz select na prijatie hodnôt z kanálov a ich výpis. Použitím príkazu select týmto spôsobom môžete súčasne prijímať údaje z viacerých kanálov a spracovávať ich hneď po prijatí.
for range
channel dokáže jednoducho prijímať údaje pomocou príkazu for range. Ak sa for range používa s kanálom, vykonáva sa vždy, keď sa do kanála pridajú údaje, a ukončí slučku, keď sa kanál zatvorí.
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}
Vyššie uvedený kód používa channel na výpis čísiel 1 a 2. V tomto kóde používame príkaz for range na prijímanie a výpis údajov vždy, keď sa do kanála pridajú údaje. A keď sa kanál zatvorí, slučku ukončíme.
Ako sme už niekoľkokrát uviedli, túto syntax možno použiť aj pre jednoduché prostriedky synchronizácie.
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}
Vyššie uvedený kód vypíše Hello, World! po 1 sekunde. V tomto kóde sme pomocou channel zmenili synchrónny kód na asynchrónny kód. Použitím channel týmto spôsobom môžete jednoducho zmeniť synchrónny kód na asynchrónny kód a nastaviť bod join.
etc
- Ak odošlete alebo prijmete údaje na kanáli nil, môžete sa dostať do nekonečnej slučky a dôjde k zablokovaniu.
- Ak odošlete údaje po zatvorení kanála, dôjde k panike.
- Aj keď kanál nie je zatvorený, GC ho uzavrie pri zbere.
mutex
spinlock
spinlock je spôsob synchronizácie, ktorý opakovane prechádza slučkou a pokúša sa získať zámok. V jazyku Go môžeme ľahko implementovať spinlock pomocou ukazovateľov.
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}
Vyššie uvedený kód je kód, ktorý implementuje balík spinlock. V tomto kóde sme použili balík sync/atomic na implementáciu SpinLock. Metóda Lock sa pokúša získať zámok pomocou atomic.CompareAndSwapUintptr a metóda Unlock uvoľní zámok pomocou atomic.StoreUintptr. Keďže táto metóda sa nepretržite pokúša získať zámok, bude naďalej využívať CPU, kým nezíska zámok, čo môže viesť k nekonečnej slučke. Preto sa spinlock odporúča používať na jednoduchú synchronizáciu alebo na krátky čas.
sync.Mutex
mutex je nástroj na synchronizáciu medzi gorutinami. mutex, ktorý poskytuje balík sync, poskytuje metódy ako Lock, Unlock, RLock, RUnlock. mutex sa dá vytvoriť pomocou sync.Mutex a na použitie zámku na čítanie/zápis sa dá použiť 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}
Vo vyššie uvedenom kóde dve gorutiny pristupujú takmer súčasne k rovnakej premennej count. Ak v tomto prípade použijeme mutex na to, aby sa kód pristupujúci k premennej count stal kritickou sekciou, môžeme zabrániť súbežnému prístupu k premennej count. Potom tento kód bude vždy vypisovať 2 bez ohľadu na to, koľkokrát sa spustí.
sync.RWMutex
sync.RWMutex je mutex, ktorý dokáže rozlišovať medzi zámkami na čítanie a zámkami na zápis. Na uzamknutie a uvoľnenie zámku na čítanie môžete použiť metódy RLock a RUnlock.
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}
Vyššie uvedený kód je kód, ktorý implementuje ConcurrentMap pomocou sync.RWMutex. V tomto kóde uzamkneme čítanie v metóde Get a uzamkneme zápis v metóde Set, aby sme mohli bezpečne pristupovať a upravovať mapu data. Dôvodom, prečo je potrebný zámok na čítanie, je to, že v prípade mnohých jednoduchých operácií čítania je možné vykonať viacero operácií čítania súčasne bez uzamknutia zápisu, ale len s uzamknutím čítania. Týmto spôsobom môžeme zlepšiť výkon tým, že uzamkneme iba čítanie, ak nedochádza k žiadnej zmene stavu a nie je potrebné uzamknúť zápis.
fakelock
fakelock je jednoduchý trik, ktorý implementuje sync.Locker. Táto štruktúra poskytuje rovnaké metódy ako sync.Mutex, ale v skutočnosti nerobí nič.
1package fakelock
2
3type FakeLock struct{}
4
5func (f *FakeLock) Lock() {}
6
7func (f *FakeLock) Unlock() {}
Vyššie uvedený kód je kód, ktorý implementuje balík fakelock. Tento balík implementuje sync.Locker a poskytuje metódy Lock a Unlock, ale v skutočnosti nerobí nič. Prečo je takýto kód potrebný, popíšeme, ak sa naskytne príležitosť.
waitgroup
sync.WaitGroup
sync.WaitGroup je nástroj, ktorý čaká, kým sa nedokončia všetky úlohy gorutín. Poskytuje metódy Add, Done a Wait. Metóda Add pridáva počet gorutín, metóda Done signalizuje, že úloha gorutiny je dokončená. A metóda Wait čaká, kým sa nedokončia všetky úlohy gorutín.
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}
Vyššie uvedený kód používa sync.WaitGroup na súčasné pridávanie hodnoty k premennej c pomocou 100 gorutín. V tomto kóde sa používa sync.WaitGroup na čakanie, kým všetky gorutiny neskončia, a potom sa vytlačí hodnota, ktorá bola pridaná k premennej c. Aj keď na jednoduché fork & join operácie stačí použiť iba kanály, použitie sync.WaitGroup je dobrou voľbou pre fork & join veľkého množstva operácií.
so slice
Ak sa používa so slicom, waitgroup môže byť vynikajúcim nástrojom na riadenie súbežných operácií bez zámkov.
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}
Vyššie uvedený kód používa iba waitgroup na súčasné generovanie 10 náhodných celých čísel každou gorutinou a ich uloženie do priradeného indexu. V tomto kóde sa používa waitgroup na čakanie, kým všetky gorutiny neskončia, a potom sa vytlačí Done. Použitím waitgroup týmto spôsobom môže viacero gorutín vykonávať operácie súčasne, ukladať dáta bez zámkov, kým všetky gorutiny neskončia, a hromadne spracovávať následné operácie po skončení operácií.
golang.org/x/sync/errgroup.ErrGroup
errgroup je balík, ktorý rozširuje sync.WaitGroup. Na rozdiel od sync.WaitGroup, ak sa počas operácie gorutiny vyskytne chyba, errgroup zruší všetky gorutiny a vráti chybu.
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}
Vyššie uvedený kód používa errgroup na vytvorenie 10 gorutín a vyvolanie chyby v 5. gorutine. Zámerne sme vyvolali chybu v piatej gorutine, aby sme ukázali prípad, keď sa vyskytne chyba. V skutočnosti by ste však mali používať errgroup na vytváranie gorutín a vykonávať rôzne následné operácie v prípade, že sa v každej gorutine vyskytne chyba.
once
Nástroj na vykonanie kódu, ktorý sa má vykonať iba raz. Súvisiaci kód je možné vykonať prostredníctvom nižšie uvedeného konštruktora.
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 jednoducho umožňuje, aby sa daná funkcia vykonala iba raz v celom rozsahu.
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}
Vyššie uvedený kód používa sync.OnceFunc na vytlačenie Hello, World!. V tomto kóde sa pomocou sync.OnceFunc vytvorí funkcia once a aj keď sa funkcia once zavolá viackrát, Hello, World! sa vytlačí iba raz.
OnceValue
OnceValue nezabezpečuje len to, že sa daná funkcia vykoná iba raz v celom rozsahu, ale aj ukladá návratovú hodnotu danej funkcie a pri opätovnom zavolaní vráti uloženú hodnotu.
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}
Vyššie uvedený kód používa sync.OnceValue na inkrementovanie premennej c o 1. V tomto kóde sa pomocou sync.OnceValue vytvorí funkcia once a aj keď sa funkcia once zavolá viackrát, vráti sa 1, pretože premenná c sa inkrementuje iba raz.
OnceValues
OnceValues funguje rovnako ako OnceValue, ale môže vrátiť viacero hodnôt.
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}
Vyššie uvedený kód používa sync.OnceValues na inkrementovanie premennej c o 1. V tomto kóde sa pomocou sync.OnceValues vytvorí funkcia once a aj keď sa funkcia once zavolá viackrát, vráti sa 1, pretože premenná c sa inkrementuje iba raz.
atomic
Balík atomic je balík, ktorý poskytuje atómové operácie. Balík atomic poskytuje metódy ako Add, CompareAndSwap, Load, Store, Swap, ale v poslednej dobe sa odporúča používať typy ako 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}
Toto je príklad, ktorý bol použitý predtým. Používa sa typ atomic.Int64 na atómové inkrementovanie premennej c. Pre inkrementovanie premennej a jej načítanie je možné použiť metódy Add a Load. Okrem toho je možné uložiť hodnotu pomocou metódy Store, vymeniť hodnotu pomocou metódy Swap a porovnať hodnotu a potom ju vymeniť, ak je vhodná, pomocou metódy CompareAndSwap.
cond
sync.Cond
Balík cond je balík, ktorý poskytuje podmienkové premenné. Balík cond je možné vytvoriť pomocou sync.Cond a poskytuje metódy 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}
Vyššie uvedený kód používa sync.Cond na čakanie, kým sa premenná ready nestane true. V tomto kóde sa pomocou sync.Cond čaká, kým sa premenná ready nestane true, a potom sa vytlačí Ready!. Použitím sync.Cond týmto spôsobom je možné zabezpečiť, aby viacero gorutín čakalo, kým sa splní určitá podmienka.
Pomocou toho je možné implementovať jednoduchý 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}
Použitím sync.Cond týmto spôsobom je možné efektívne čakať a opäť fungovať, keď je splnená podmienka, namiesto používania veľkého množstva CPU s spin-lock.
semaphore
golang.org/x/sync/semaphore.Semaphore
Balík semaphore je balík, ktorý poskytuje semafory. Balík semaphore je možné vytvoriť pomocou golang.org/x/sync/semaphore.Semaphore a poskytuje metódy 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}
Vyššie uvedený kód používa semaphore na vytvorenie semaforu a semafor sa používa na získanie semaforu pomocou metódy Acquire a uvoľnenie semaforu pomocou metódy Release. V tomto kóde sme ukázali, ako pomocou semaphore získať a uvoľniť semafor.
Záver
Zdá sa, že základné veci sú tu pokryté. Dúfam, že na základe obsahu tohto článku chápete, ako používať gorutiny na riadenie súbežnosti, a budete ich môcť skutočne používať. Dúfam, že vám tento článok pomohol. Ďakujem.