GoSudaPachetul de pornire pentru concurență în Go
Prezentare generală
Scurtă introducere
Limbajul Go dispune de multe instrumente pentru gestionarea concurenței. În acest articol vom prezenta o parte dintre acestea și câteva trucuri.
Goroutine?
goroutine este un nou model de concurență suportat de limbajul Go. În mod obișnuit, un program primește de la sistemul de operare fire de execuție OS pentru a efectua mai multe sarcini simultan, executând sarcinile în paralel, în număr egal cu numărul de nuclee. Pentru a realiza concurența la o scară mai mică, se creează în spațiul utilizatorului (userland) fire verzi (green threads), astfel încât mai multe fire verzi rulează pe rând în cadrul unui singur fir OS. Goroutine-urile reprezintă o formă de fire verzi mai mică și mai eficientă. Ele utilizează mai puțină memorie decât firele de execuție și pot fi create și schimbate mai rapid decât acestea.
Pentru a utiliza goroutine, este suficient să folosiți cuvântul cheie go. Acest lucru permite ca, în procesul scrierii programului, codul sincron să fie executat intuitiv ca cod asincron.
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}
Acest cod transformă un cod sincron, care așteaptă 1 secundă și apoi afișează Hello, World!, într-un flux asincron. Exemplul este simplu, dar atunci când un cod mai complex este transformat din sincron în asincron, lizibilitatea, vizibilitatea și înțelegerea codului devin mai bune decât cu metode precum async await sau promise.
Totuși, în multe cazuri, dacă nu se înțeleg fluxul de apelare asincronă a unui cod sincron și fluxuri precum fork & join (un flux asemănător cu divide et impera), pot rezulta goroutine de calitate slabă. Pentru a face față unor astfel de cazuri, în acest articol vom prezenta câteva metode și tehnici.
Gestionarea concurenței
context
Este posibil să fie surprinzător că prima tehnică prezentată este context. Totuși, în limbajul Go, context depășește simpla funcționalitate de anulare și joacă un rol excelent în gestionarea întregului arbore de sarcini. Pentru cei care nu cunosc acest pachet, îl voi descrie pe scurt.
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}
Codul de mai sus utilizează context pentru a afișa Context is done! după 1 secundă. context permite verificarea anulării prin metoda Done() și oferă diverse metode de anulare, precum WithCancel, WithTimeout, WithDeadline și WithValue.
Să construim un exemplu simplu. Să presupunem că scrieți un cod care utilizează modelul aggregator pentru a obține user, post și comment, iar toate cererile trebuie să fie efectuate în decurs de 2 secunde. Atunci puteți scrie astfel:
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}
Codul de mai sus afișează Timeout! dacă nu reușește să obțină toate datele în decurs de 2 secunde și afișează All data is fetched! dacă le obține pe toate. În acest mod, folosind context, anularea și expirarea timpului pot fi gestionate cu ușurință chiar și în cadrul unui cod în care rulează mai multe goroutine.
Diversele funcții și metode legate de context pot fi consultate pe godoc context. Sper că, după ce veți studia cele simple, le veți putea utiliza cu ușurință.
channel
unbuffered channel
channel este un instrument pentru comunicarea între goroutine. channel poate fi creat cu make(chan T), unde T este tipul de date pe care îl va transmite acel channel. Datele se trimit și se primesc cu <-, iar channel poate fi închis cu 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}
Codul de mai sus utilizează channel pentru a afișa 1 și 2. Aici se demonstrează doar trimiterea și primirea simplă a valorilor prin channel. Totuși, channel oferă mai multe funcționalități decât atât. Mai întâi, să analizăm buffered channel și unbuffered channel. Exemplul de mai sus este un unbuffered channel, în care acțiunea de trimitere a datelor în canal și acțiunea de primire a acestora trebuie să aibă loc simultan. Dacă aceste acțiuni nu au loc simultan, poate apărea un deadlock.
buffered channel
Ce se întâmplă dacă, în loc de o simplă afișare, codul de mai sus ar reprezenta două procese care efectuează sarcini grele? Dacă al doilea proces, care citește și procesează datele, rămâne blocat o perioadă lungă, atunci și primul proces se va opri pe durata respectivă. Pentru a preveni această situație, putem utiliza 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}
Codul de mai sus utilizează buffered channel pentru a afișa 1 și 2. În acest cod, buffered channel face ca acțiunea de trimitere a datelor în canal și acțiunea de primire a acestora să nu fie obligatoriu simultane. Atunci când canalul are un buffer, există un spațiu liber egal cu lungimea acestuia, ceea ce permite prevenirea întârzierii sarcinilor cauzate de influența sarcinilor ulterioare.
select
Atunci când lucrați cu mai multe canale, puteți implementa cu ușurință o structură fan-in folosind 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}
Codul de mai sus creează 3 canale care transmit periodic valorile 1, 2 și 3, și utilizează select pentru a primi valori din canale și a le afișa. În acest mod, folosind select, se pot primi date simultan din mai multe canale și le puteți procesa pe măsură ce sosesc.
for range
channel poate fi utilizat cu for range pentru a primi datele cu ușurință. Atunci când for range este folosit pe un canal, bucla se execută de fiecare dată când se adaugă date în canal și se încheie când canalul este închis.
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}
Codul de mai sus utilizează channel pentru a afișa 1 și 2. În acest cod, for range este utilizat pentru a primi și afișa datele de fiecare dată când sunt adăugate în canal. Când canalul este închis, bucla se încheie.
După cum s-a arătat de câteva ori mai sus, această sintaxă poate fi utilizată și ca simplu mecanism de sincronizare.
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}
Codul de mai sus așteaptă 1 secundă și apoi afișează Hello, World!. În acest cod, channel este utilizat pentru a transforma codul sincron în cod asincron. În acest mod, folosind channel, codul sincron poate fi transformat ușor în cod asincron, iar un punct de join poate fi stabilit.
etc
- Dacă se trimit sau se primesc date pe un nil channel, se poate intra într-o buclă infinită, ceea ce poate duce la deadlock.
- Dacă se trimit date după ce canalul a fost închis, apare un panic.
- Chiar dacă nu închideți canalul în mod explicit, GC îl va colecta și îl va închide.
mutex
spinlock
spinlock este o metodă de sincronizare care încearcă în mod repetat să obțină blocarea, parcurgând o buclă. În limbajul Go, un spinlock poate fi implementat cu ușurință folosind pointeri.
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}
Codul de mai sus este implementarea pachetului spinlock. În acest cod, SpinLock este implementat folosind pachetul sync/atomic. Metoda Lock încearcă să obțină blocarea folosind atomic.CompareAndSwapUintptr, iar metoda Unlock eliberează blocarea folosind atomic.StoreUintptr. Deoarece această metodă încearcă neîntrerupt să obțină blocarea, procesorul este utilizat continuu până când blocarea este obținută, ceea ce poate duce la o buclă infinită. Prin urmare, spinlock este recomandat pentru sincronizări simple sau pentru utilizare doar pe perioade scurte de timp.
sync.Mutex
mutex este un instrument pentru sincronizarea între goroutine. mutex-ul furnizat de pachetul sync oferă metode precum Lock, Unlock, RLock și RUnlock. mutex poate fi creat cu sync.Mutex, iar pentru blocarea de citire/scriere se poate utiliza 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}
În codul de mai sus, două goroutine accesează aproape simultan aceeași variabilă count. Dacă folosim mutex pentru a transforma codul care accesează variabila count într-o secțiune critică, putem preveni accesul concurent la aceasta. Astfel, acest cod va afișa întotdeauna 2, indiferent de numărul de rulări.
sync.RWMutex
sync.RWMutex este un mutex care permite utilizarea separată a blocărilor de citire și de scriere. Cu ajutorul metodelor RLock și RUnlock se pot seta și elibera blocările de citire.
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}
Codul de mai sus implementează ConcurrentMap folosind sync.RWMutex. În acest cod, metoda Get setează o blocare de citire, iar metoda Set setează o blocare de scriere, astfel încât harta data poate fi accesată și modificată în siguranță. Motivul pentru care se utilizează blocarea de citire este că, în cazul în care există multe operații simple de citire, se permite ca mai multe goroutine să efectueze simultan operații de citire, fără a bloca și scrierea. Astfel, în situațiile în care nu există modificări de stare și nu este necesară blocarea de scriere, performanța poate fi îmbunătățită utilizând doar blocarea de citire.
fakelock
fakelock este un truc simplu de implementare a sync.Locker. Această structură oferă aceleași metode ca sync.Mutex, dar nu efectuează nicio acțiune reală.
1package fakelock
2
3type FakeLock struct{}
4
5func (f *FakeLock) Lock() {}
6
7func (f *FakeLock) Unlock() {}
Codul de mai sus este implementarea pachetului fakelock. Acest pachet implementează sync.Locker și oferă metodele Lock și Unlock, dar în realitate nu efectuează nicio operație. Voi explica mai târziu, dacă va fi ocazia, de ce este necesar un astfel de cod.
waitgroup
sync.WaitGroup
sync.WaitGroup este un instrument care așteaptă până când toate sarcinile goroutine-urilor sunt finalizate. Oferă metodele Add, Done și Wait: prin Add se adaugă numărul de goroutine, prin Done se anunță că o goroutine și-a încheiat sarcina, iar prin Wait se așteaptă până când toate goroutine-urile își termină sarcinile.
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}
Codul de mai sus utilizează sync.WaitGroup pentru ca 100 de goroutine să adauge simultan valori variabilei c. În acest cod, sync.WaitGroup este utilizat pentru a aștepta finalizarea tuturor goroutine-urilor, după care se afișează valoarea acumulată în c. Pentru un fork & join al unui număr mic de sarcini, este suficient să utilizați doar canale, dar pentru un fork & join al unui număr mare de sarcini, utilizarea sync.WaitGroup este o alegere bună.
with slice
Atunci când este utilizat împreună cu un slice, waitgroup poate deveni un instrument excelent pentru gestionarea sarcinilor executate concurent, fără blocări.
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}
Codul de mai sus utilizează doar waitgroup pentru ca fiecare goroutine să genereze simultan câte un număr întreg aleatoriu și să îl stocheze la indexul atribuit. În acest cod, waitgroup este utilizat pentru a aștepta finalizarea tuturor goroutine-urilor, după care se afișează Done. În acest mod, folosind waitgroup, mai multe goroutine pot efectua sarcini simultan și pot stoca datele fără blocări, iar după finalizarea tuturor goroutine-urilor se poate efectua centralizat o prelucrare finală.
golang.org/x/sync/errgroup.ErrGroup
errgroup este un pachet care extinde sync.WaitGroup. Spre deosebire de sync.WaitGroup, errgroup anulează toate goroutine-urile și returnează eroarea dacă apare o eroare în una dintre sarcini.
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}
Codul de mai sus utilizează errgroup pentru a crea 10 goroutine, iar în a cincea goroutine se generează o eroare. Am generat intenționat o eroare în a cincea goroutine pentru a arăta cazul în care apare o eroare. În utilizarea reală, errgroup se folosește pentru a crea goroutine și pentru a efectua diverse prelucrări ulterioare în cazul în care apare o eroare în oricare dintre ele.
once
Este un instrument pentru executarea codului care trebuie să ruleze o singură dată. Codul respectiv poate fi executat prin constructorii de mai jos.
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 permite pur și simplu ca funcția respectivă să fie executată exact o singură dată, pe întreaga durată.
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}
Codul de mai sus utilizează sync.OnceFunc pentru a afișa Hello, World!. În acest cod, sync.OnceFunc este utilizat pentru a crea funcția once, iar chiar dacă aceasta este apelată de mai multe ori, Hello, World! va fi afișat o singură dată.
OnceValue
OnceValue nu înseamnă doar că funcția respectivă este executată exact o singură dată pe întreaga durată, ci și că valoarea returnată de funcție este stocată, iar la apelurile ulterioare se returnează valoarea salvată.
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}
Codul de mai sus utilizează sync.OnceValue pentru a incrementa variabila c cu 1. În acest cod, sync.OnceValue este utilizat pentru a crea funcția once, iar chiar dacă aceasta este apelată de mai multe ori, variabila c este incrementată o singură dată și funcția returnează valoarea 1.
OnceValues
OnceValues funcționează în același mod ca OnceValue, dar poate returna mai multe valori.
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}
Codul de mai sus utilizează sync.OnceValues pentru a incrementa variabila c cu 1. În acest cod, sync.OnceValues este utilizat pentru a crea funcția once, iar chiar dacă aceasta este apelată de mai multe ori, variabila c este incrementată o singură dată și funcția returnează valorile corespunzătoare lui 1.
atomic
Pachetul atomic oferă operații atomice. Acesta oferă metode precum Add, CompareAndSwap, Load, Store și Swap, dar în ultima vreme se recomandă utilizarea tipurilor precum Int64, Uint64 și 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}
Este exemplul utilizat anterior. Codul folosește tipul atomic.Int64 pentru a incrementa atomic variabila c. Metodele Add și Load permit incrementarea atomică a variabilei și citirea acesteia. De asemenea, metoda Store permite stocarea unei valori, metoda Swap permite înlocuirea valorii, iar metoda CompareAndSwap permite compararea valorii și, dacă aceasta corespunde, înlocuirea ei.
cond
sync.Cond
Pachetul cond oferă variabile de condiție. Acesta poate fi creat cu sync.Cond și oferă metodele 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}
Codul de mai sus utilizează sync.Cond pentru a aștepta până când variabila ready devine true. În acest cod, sync.Cond este utilizat pentru a aștepta până când variabila ready devine true, după care se afișează Ready!. În acest mod, folosind sync.Cond, mai multe goroutine pot aștepta până când este îndeplinită o anumită condiție.
Folosind acest mecanism, putem implementa o queue simplă.
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}
În acest mod, folosind sync.Cond, în loc să consumăm mult CPU cu spin-lock, putem aștepta eficient și putem relua activitatea atunci când condiția este îndeplinită.
semaphore
golang.org/x/sync/semaphore.Semaphore
Pachetul semaphore oferă semafoare. Acesta poate fi creat cu golang.org/x/sync/semaphore.Semaphore și oferă metodele 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}
Codul de mai sus creează un semafor folosind semaphore, apoi utilizează metoda TryAcquire pentru a încerca obținerea acestuia și metoda Release pentru a-l elibera. În acest cod, este prezentat modul în care semaforul poate fi obținut și eliberat.
Încheiere
Conținutul de bază ar trebui să fie suficient până aici. Pe baza acestui articol, sper că veți înțelege cum să gestionați concurența folosind goroutine și că veți putea aplica aceste cunoștințe în practică. Sper că acest articol v-a fost util. Mulțumesc.