GoSudaGo-samtidighedsstarterpakke
Oversigt
Kort introduktion
Go-sproget har mange værktøjer til håndtering af samtidighed. I denne artikel vil jeg præsentere nogle af dem samt nogle tricks.
Goroutines?
goroutine er en ny form for samtidighedsmodel, der understøttes af Go-sproget. Normalt modtager et program OS-tråde fra operativsystemet for at udføre flere opgaver samtidigt, og udfører opgaver parallelt svarende til antallet af kerner. For at udføre samtidighed i mindre skala oprettes grønne tråde i user space, således at flere grønne tråde kører skiftevis inden for én OS-tråd. Goroutines er dog gjort endnu mindre og mere effektive end sådanne grønne tråde. Disse goroutines bruger mindre hukommelse end tråde og kan oprettes og skiftes hurtigere end tråde.
For at bruge goroutines er det tilstrækkeligt blot at anvende nøgleordet go. Dette gør det muligt intuitivt at køre synkron kode som asynkron kode under programmering.
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}
Denne kode ændrer en synkron kode, der holder pause i 1 sekund og derefter udskriver Hello, World!, til et asynkront flow. Selvom dette eksempel er simpelt, bliver læsbarhed, synlighed og forståelighed bedre, når man ændrer mere komplekse kodestykker fra synkron til asynkron kode, end med eksisterende metoder som async await eller promise.
Dog opstår der i mange tilfælde dårlig goroutine-kode, hvis man ikke forstår flowet ved blot at kalde synkron kode asynkront, og flows som fork & join (som ligner divide and conquer). For at forberede sig på sådanne tilfælde vil jeg i denne artikel præsentere nogle metoder og teknikker.
Håndtering af samtidighed
context
Det kan være overraskende, at context optræder som den første håndteringsteknik. Men i Go-sproget spiller context en fremragende rolle i at administrere hele opgavetræet ud over en simpel annulleringsfunktion. For dem, der ikke kender pakken, vil jeg kort forklare den.
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}
Ovenstående kode bruger context til at udskrive Context is done! efter 1 sekund. Man kan kontrollere, om der er annulleret, via metoden Done() på context, og den tilbyder forskellige annulleringsmetoder gennem metoder som WithCancel, WithTimeout, WithDeadline og WithValue.
Lad os lave et simpelt eksempel. Antag, at man for at hente data skriver kode, der bruger aggregator-mønsteret til at hente user, post og comment, og at alle anmodninger skal være gennemført inden for 2 sekunder. Det kan så skrives som følger.
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}
Ovenstående kode udskriver Timeout!, hvis alle data ikke er hentet inden for 2 sekunder, og All data is fetched!, hvis alle data hentes. På denne måde kan man med context nemt håndtere annullering og timeout, selv i kode hvor flere goroutines kører.
Forskellige context-relaterede funktioner og metoder kan findes på godoc context. Jeg håber, at du lærer det grundlæggende og kan bruge det med lethed.
channel
unbuffered channel
channel er et værktøj til kommunikation mellem goroutines. En channel kan oprettes med make(chan T). Her er T den datatype, som kanalen skal overføre. Man kan sende og modtage data med <-, og lukke en channel med 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}
Ovenstående kode bruger channel til at udskrive 1 og 2. Denne kode viser blot at sende og modtage værdier via en kanal. Men channel tilbyder flere funktioner end dette. Lad os først se på buffered channel og unbuffered channel. Inden vi begynder, skal det bemærkes, at eksemplet ovenfor er en unbuffered channel, hvor handlingen med at sende data til kanalen og handlingen med at modtage data skal ske samtidigt. Hvis disse handlinger ikke sker samtidigt, kan der opstå deadlock.
buffered channel
Hvad hvis den ovenstående kode ikke blot udskriver, men er to processer, der udfører tunge opgaver? Hvis den anden proces læser og behandler, men bruger lang tid, vil den første proces også stoppe i så lang tid. For at forhindre denne situation kan vi bruge 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}
Ovenstående kode bruger buffered channel til at udskrive 1 og 2. I denne kode bruges buffered channel, så handlingen med at sende data til kanalen og handlingen med at modtage data ikke behøver at ske samtidigt. Når der er en buffer i kanalen, opstår der plads svarende til bufferlængden, og dermed kan forsinkelser forårsaget af påvirkning fra efterfølgende opgaver forhindres.
select
Når man arbejder med flere kanaler, kan man let implementere en fan-in-struktur ved hjælp af select-syntaksen.
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}
Ovenstående kode opretter tre kanaler, der periodisk sender 1, 2 og 3, og bruger select til at modtage værdier fra kanalerne og udskrive dem. På denne måde kan man med select modtage data fra flere kanaler samtidigt og behandle dem i takt med, at værdierne modtages.
for range
channel kan nemt modtage data med for range. Når for range bruges på en kanal, udføres loopet hver gang der tilføjes data til kanalen, og loopet afsluttes, når kanalen lukkes.
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}
Ovenstående kode bruger channel til at udskrive 1 og 2. I denne kode bruges for range til at modtage og udskrive data, hver gang der tilføjes data til kanalen. Når kanalen lukkes, afsluttes loopet.
Som tidligere vist kan denne syntaks også bruges som et simpelt synkroniseringsmiddel.
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}
Ovenstående kode holder pause i 1 sekund og udskriver derefter Hello, World!. I denne kode bruges channel til at ændre synkron kode til asynkron kode. På denne måde kan man med channel nemt ændre synkron kode til asynkron kode og etablere et join-punkt.
etc
- Hvis man sender eller modtager data på en nil channel, kan der opstå en uendelig løkke og deadlock.
- Hvis man sender data efter at have lukket en kanal, opstår der en panic.
- Selv hvis man ikke lukker en kanal eksplicit, lukker GC den, når den indsamler den.
mutex
spinlock
spinlock er en synkroniseringsmetode, der gentagne gange i en løkke forsøger at tage låsen. I Go-sproget kan man let prøve at implementere en spinlock ved hjælp af pointere.
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}
Ovenstående kode implementerer pakken spinlock. Denne kode implementerer SpinLock ved hjælp af pakken sync/atomic. I metoden Lock forsøges låsen med atomic.CompareAndSwapUintptr, og i metoden Unlock frigives låsen med atomic.StoreUintptr. Da denne metode forsøger at tage låsen uden pause, bruger den CPU kontinuerligt, indtil låsen opnås, og den kan ende i en uendelig løkke. Derfor er det bedst at bruge spinlock til simpel synkronisering eller til tilfælde, hvor den kun bruges i kort tid.
sync.Mutex
mutex er et værktøj til synkronisering mellem goroutines. mutex, som leveres af pakken sync, tilbyder metoder som Lock, Unlock, RLock og RUnlock. En mutex kan oprettes med sync.Mutex, og man kan også bruge en læse/skrive-lås med 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}
I ovenstående kode får to goroutines næsten samtidigt adgang til den samme variabel count. Hvis man gør koden, der tilgår variablen count, til et kritisk område ved hjælp af mutex, kan samtidig adgang til count forhindres. Så vil denne kode uanset antal kørsler altid udskrive 2.
sync.RWMutex
sync.RWMutex er en mutex, hvor læselåse og skrivelåse kan bruges adskilt. Man kan tage og frigive læselåsen med metoderne RLock og 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}
Ovenstående kode implementerer ConcurrentMap ved hjælp af sync.RWMutex. I denne kode tager metoden Get en læselås, og metoden Set tager en skrivelås, så data-mappet kan tilgås og ændres sikkert. Læselåsen er nødvendig, fordi man i tilfælde med mange simple læseoperationer ikke tager en skrivelås, men kun en læselås, så flere goroutines kan udføre læseoperationer samtidigt. Dermed kan man forbedre ydeevnen ved kun at tage en læselås i de tilfælde, hvor der ikke er behov for en skrivelås, fordi tilstanden ikke ændres.
fakelock
fakelock er et simpelt trick, der implementerer sync.Locker. Denne struct tilbyder de samme metoder som sync.Mutex, men udfører faktisk ingen handling.
1package fakelock
2
3type FakeLock struct{}
4
5func (f *FakeLock) Lock() {}
6
7func (f *FakeLock) Unlock() {}
Ovenstående kode implementerer pakken fakelock. Pakken implementerer sync.Locker og tilbyder metoderne Lock og Unlock, men gør faktisk ingenting. Jeg vil forklare, hvorfor denne kode er nødvendig, når der er mulighed for det.
waitgroup
sync.WaitGroup
sync.WaitGroup er et værktøj, der venter, indtil alle goroutines opgaver er afsluttet. Den tilbyder metoderne Add, Done og Wait. Med metoden Add tilføjes antallet af goroutines, med metoden Done meddeles det, at en goroutines opgave er afsluttet, og med metoden Wait venter man, indtil alle goroutines opgaver er afsluttet.
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}
Ovenstående kode bruger sync.WaitGroup til at lade 100 goroutines samtidigt lægge værdi til variablen c. I denne kode venter sync.WaitGroup på, at alle goroutines er færdige, og derefter udskrives den værdi, der er lagt til c. Når man blot laver fork & join af nogle få opgaver, er det tilstrækkeligt kun at bruge kanaler, men ved fork & join af mange opgaver er sync.WaitGroup også et godt valg.
with slice
Når det bruges sammen med slices, kan waitgroup blive et fremragende værktøj til at håndtere samtidig udførelse uden lås.
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}
Ovenstående kode bruger udelukkende waitgroup, hvor hver goroutine samtidigt genererer et tilfældigt heltal mellem 0 og 99 og gemmer det på det tildelte indeks. I denne kode venter waitgroup på, at alle goroutines er færdige, og udskriver derefter Done. På denne måde kan man med waitgroup lade flere goroutines udføre opgaver samtidigt, gemme data uden lås, indtil alle goroutines er færdige, og derefter udføre efterbehandling samlet.
golang.org/x/sync/errgroup.ErrGroup
errgroup er en pakke, der udvider sync.WaitGroup. I modsætning til sync.WaitGroup annullerer errgroup alle goroutines og returnerer fejlen, hvis en fejl opstår i en af goroutines opgaver.
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}
Ovenstående kode bruger errgroup til at oprette 10 goroutines og får den 5. goroutine til at give en fejl. Jeg har med vilje fremkaldt en fejl i den femte goroutine for at vise tilfældet, hvor der opstår en fejl. I praksis bruger man errgroup til at oprette goroutines og udføre forskellig efterbehandling i tilfælde, hvor der opstår fejl i de enkelte goroutines.
once
Det er et værktøj til at udføre kode, der kun må køres én gang. Man kan køre den relaterede kode gennem følgende konstruktører.
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 sørger blot for, at den pågældende funktion kun kan køres præcis én gang i hele programmet.
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}
Ovenstående kode bruger sync.OnceFunc til at udskrive Hello, World!. I denne kode oprettes funktionen once med sync.OnceFunc, og selvom once kaldes flere gange, udskrives Hello, World! kun én gang.
OnceValue
OnceValue sørger ikke blot for, at den pågældende funktion kun køres præcis én gang, men gemmer også funktionens returværdi, så den gemte værdi returneres ved de efterfølgende kald.
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}
Ovenstående kode bruger sync.OnceValue til at øge variablen c med 1. I denne kode oprettes funktionen once med sync.OnceValue, og selvom once kaldes flere gange, returnerer den 1, fordi variablen c kun er blevet øget én gang.
OnceValues
OnceValues fungerer på samme måde som OnceValue, men kan returnere flere værdier.
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}
Ovenstående kode bruger sync.OnceValues til at øge variablen c med 1. I denne kode oprettes funktionen once med sync.OnceValues, og selvom once kaldes flere gange, returnerer den 1, fordi variablen c kun er blevet øget én gang.
atomic
Pakken atomic er en pakke, der tilbyder atomare operationer. Pakken atomic tilbyder metoder som Add, CompareAndSwap, Load, Store og Swap, men for nylig anbefales det at bruge typer som Int64, Uint64 og 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}
Dette er et eksempel, der blev brugt tidligere. Koden bruger typen atomic.Int64 til at øge variablen c atomisk. Med metoderne Add og Load kan variablen øges atomisk og læses. Desuden kan man gemme en værdi med metoden Store, udskifte en værdi med metoden Swap, og med metoden CompareAndSwap sammenligne en værdi og udskifte den, hvis den passer.
cond
sync.Cond
Pakken cond er en pakke, der tilbyder betingelsesvariabler. Pakken cond kan oprettes med sync.Cond, og tilbyder metoderne Wait, Signal og 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}
Ovenstående kode bruger sync.Cond til at vente, indtil variablen ready bliver true. I denne kode venter sync.Cond, indtil variablen ready bliver true, og derefter udskrives Ready!. På denne måde kan man med sync.Cond få flere goroutines til at vente, indtil en bestemt betingelse er opfyldt.
Ved at udnytte dette kan man implementere en simpel 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}
Når man bruger sync.Cond på denne måde, kan man vente effektivt i stedet for at bruge stort CPU-forbrug med en spin-lock, og fortsætte, når betingelsen er opfyldt.
semaphore
golang.org/x/sync/semaphore.Semaphore
Pakken semaphore er en pakke, der tilbyder semaforer. Pakken semaphore kan oprettes med golang.org/x/sync/semaphore.Semaphore, og tilbyder metoderne Acquire, Release og 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}
Ovenstående kode opretter en semafor med semaphore, henter semaforen med metoden Acquire og frigiver den med metoden Release. I denne kode er det vist, hvordan man henter og frigiver en semafor med semaphore.
Afslutning
Det grundlæggende indhold burde være tilstrækkeligt her. Ud fra indholdet i denne artikel håber jeg, at du kan forstå, hvordan man håndterer samtidighed med goroutines, og at du kan bruge det i praksis. Jeg håber, at denne artikel har været nyttig for dig. Tak.