GoSudaGo-samtidighetsstarterpakke
Oversikt
Kort introduksjon
Go-språket har mange verktøy for håndtering av samtidighet. I denne artikkelen vil jeg presentere noen av dem sammen med noen triks.
Goroutiner?
goroutine er en ny form for samtidighetsmodell som støttes av Go-språket. Vanligvis får et program OS-tråder fra operativsystemet for å utføre flere oppgaver samtidig, og utfører oppgaver parallelt tilsvarende antall kjerner. For å oppnå samtidighet på et mindre nivå opprettes grønne tråder i brukerrommet, slik at flere grønne tråder kjører oppgaver etter tur innenfor én OS-tråd. Goroutiner er imidlertid laget som en mer kompakt og effektiv variant av slike grønne tråder. Goroutiner bruker mindre minne enn tråder og kan opprettes og byttes raskere enn tråder.
For å bruke goroutiner er det tilstrekkelig å bruke nøkkelordet go. Dette gjør det intuitivt mulig å kjø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 koden endrer en synkron kode, som venter ett sekund og deretter skriver ut Hello, World!, til en asynkron flyt. Selv om eksemplet nå er enkelt, blir lesbarhet, synlighet og forståelighet bedre enn med metoder som async await eller promise når man konverterer en mer kompleks kode fra synkron til asynkron.
I mange tilfeller kan dårlig goroutine-kode dog oppstå når man ikke forstår flyten der synkron kode bare kalles asynkront, eller flyter som fork & join (som ligner på divide and conquer). For å håndtere slike tilfeller vil jeg i denne artikkelen presentere noen metoder og teknikker.
Håndtering av samtidighet
context
Det kan være overraskende at context dukker opp som den første håndteringsteknikken. Men i Go-språket spiller context en fremragende rolle i å håndtere hele arbeidstreet, utover en enkel avbrytelsesfunksjon. For dem som ikke kjenner til det, vil jeg kort forklare pakken.
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}
Koden ovenfor bruker context til å skrive ut Context is done! etter ett sekund. context kan sjekke om avbrytelse har skjedd via metoden Done(), og tilbyr ulike avbrytelsesmetoder via metoder som WithCancel, WithTimeout, WithDeadline og WithValue.
La oss lage et enkelt eksempel. Anta at du skriver kode som bruker aggregator-mønsteret for å hente user, post og comment, og at alle forespørsler må fullføres innen to sekunder. Da kan du skrive følgende.
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}
Koden ovenfor skriver ut Timeout! hvis ikke alle data hentes innen to sekunder, og skriver ut All data is fetched! hvis alle data hentes. På denne måten kan du enkelt håndtere avbrytelse og tidsavbrudd i kode der flere goroutiner kjører, når du bruker context.
Ulike funksjoner og metoder knyttet til context finnes på godoc context. Jeg håper at du kan lære det enkle og bruke det med letthet.
channel
unbuffered channel
channel er et verktøy for kommunikasjon mellom goroutiner. En channel kan opprettes med make(chan T), der T er datatypen som kanalen skal overføre. Data sendes og mottas med <-, og channel lukkes 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}
Koden ovenfor bruker channel til å skrive ut 1 og 2. Dette eksemplet viser bare enkel sending og mottak av verdier i en kanal, men channel tilbyr flere funksjoner enn dette. Først skal vi se på buffered channel og unbuffered channel. Før vi begynner, må det nevnes at eksemplet ovenfor bruker en unbuffered channel, der handlingen med å sende data til kanalen og handlingen med å motta data må skje samtidig. Hvis disse handlingene ikke skjer samtidig, kan en deadlock oppstå.
buffered channel
Hva om koden ovenfor ikke bare skriver ut, men utfører to prosesser med tung oppgave? Hvis den andre prosessen leser og behandler data, og bruker lang tid på dette, vil også den første prosessen stoppe i samme tidsrom. For å unngå denne situasjonen kan vi bruke 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}
Koden ovenfor bruker buffered channel til å skrive ut 1 og 2. I denne koden trenger ikke handlingen med å sende data til channel og handlingen med å motta data å skje samtidig, fordi vi har laget en buffer. Når kanalen har en buffer, får den kapasitet tilsvarende bufferlengden, og dette kan forhindre forsinkelser i arbeid som skyldes påvirkning fra oppgaver med lavere prioritet.
select
Når du arbeider med flere kanaler, kan du enkelt implementere en fan-in-struktur ved hjelp av 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}
Koden ovenfor oppretter tre kanaler som periodisk sender 1, 2 og 3, og bruker select til å motta og skrive ut verdier fra kanalene. På denne måten kan select motta data samtidig fra flere kanaler og behandle verdiene i den rekkefølgen de kommer.
for range
channel kan enkelt motta data ved hjelp av for range. Når for range brukes på en kanal, kjøres løkken hver gang det legges til data i kanalen, og løkken avsluttes 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}
Koden ovenfor bruker channel til å skrive ut 1 og 2. I denne koden bruker vi for range til å motta og skrive ut data hver gang det legges til data i kanalen. Når kanalen lukkes, avsluttes løkken.
Som vist flere ganger tidligere kan denne syntaksen også brukes som et enkelt synkroniseringsverktøy.
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}
Koden ovenfor venter ett sekund og skriver deretter ut Hello, World!. I denne koden brukes channel til å endre synkron kode til asynkron kode. På denne måten kan du enkelt endre synkron kode til asynkron kode med channel og angi et join-punkt.
etc
- Hvis du sender data til eller mottar data fra en nil channel, kan det oppstå en uendelig løkke som fører til deadlock.
- Hvis du sender data etter at en kanal er lukket, oppstår en panic.
- Selv om du ikke lukker en kanal eksplisitt, vil GC samle den inn og lukke den.
mutex
spinlock
spinlock er en synkroniseringsmetode som gjentatte ganger prøver å ta låsen i en løkke. I Go-språket kan du enkelt prøve å implementere en spinlock ved hjelp av pekere.
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}
Koden ovenfor implementerer pakken spinlock. Her er SpinLock implementert ved hjelp av pakken sync/atomic. I metoden Lock brukes atomic.CompareAndSwapUintptr til å forsøke å ta låsen, og i metoden Unlock brukes atomic.StoreUintptr til å frigjøre låsen. Siden denne metoden prøver å ta låsen uten pause, bruker den CPU kontinuerlig til låsen er tatt, og kan dermed henge seg i en uendelig løkke. Derfor bør spinlock brukes til enkel synkronisering eller bare i korte tidsrom.
sync.Mutex
mutex er et verktøy for synkronisering mellom goroutiner. mutex fra pakken sync tilbyr metodene Lock, Unlock, RLock og RUnlock. En mutex kan opprettes med sync.Mutex, og du kan også bruke en lese-/skrivelå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 koden ovenfor får to goroutiner nesten samtidig tilgang til den samme variabelen count. Hvis du bruker mutex for å gjøre koden som får tilgang til count til en kritisk seksjon, kan du hindre samtidig tilgang til count. Da vil denne koden alltid skrive ut 2, uansett hvor mange ganger den kjøres.
sync.RWMutex
sync.RWMutex er en mutex der lese- og skrivelås kan brukes separat. Du kan ta og frigjøre leselåsen med metodene 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}
Koden ovenfor implementerer ConcurrentMap ved hjelp av sync.RWMutex. I denne koden tar metoden Get leselåsen, og metoden Set tar skrivelåsen, slik at kartet data kan nås og endres trygt. Grunnen til at leselås er nødvendig, er at når det er mange rene leseoperasjoner, kan flere goroutiner utføre lesing samtidig ved bare å ta leselåsen, uten skrivelås. På denne måten kan ytelsen forbedres ved å bare ta leselåsen i tilfeller der det ikke er nødvendig å endre tilstanden og skrivelås derfor ikke trengs.
fakelock
fakelock er et enkelt triks som implementerer sync.Locker. Denne strukturen tilbyr de samme metodene som sync.Mutex, men utfører ingen faktisk handling.
1package fakelock
2
3type FakeLock struct{}
4
5func (f *FakeLock) Lock() {}
6
7func (f *FakeLock) Unlock() {}
Koden ovenfor implementerer pakken fakelock. Pakken implementerer sync.Locker og tilbyr metodene Lock og Unlock, men den gjør faktisk ingenting. Hvorfor en slik kode er nødvendig, vil jeg forklare ved en senere anledning.
waitgroup
sync.WaitGroup
sync.WaitGroup er et verktøy som venter til alle oppgavene til goroutinene er fullført. Den tilbyr metodene Add, Done og Wait. Med metoden Add legger du til antall goroutiner, med Done melder du at en goroutine er ferdig, og med Wait venter du til alle goroutinene er ferdige.
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}
Koden ovenfor bruker sync.WaitGroup til å la 100 goroutiner samtidig legge til verdier i variabelen c. I denne koden venter sync.WaitGroup til alle goroutinene er ferdige, og deretter skrives verdien ut som er lagt til i c. Når du bare skal utføre fork & join på noen få oppgaver, er det tilstrekkelig å bruke kanaler alene, men når du skal utføre fork & join på et stort antall oppgaver, er sync.WaitGroup også et godt valg.
with slice
Når den brukes sammen med en slice, kan waitgroup bli et utmerket verktøy for å håndtere samtidig utførelse uten 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}
Koden ovenfor bruker kun waitgroup, der hver goroutine samtidig genererer ett av 10 tilfeldige heltall og lagrer det på den tildelte indeksen. I denne koden venter waitgroup til alle goroutinene er ferdige, og deretter skrives Done ut. På denne måten kan flere goroutiner utføre oppgaver samtidig, lagre data uten lås mens alle goroutinene kjører, og utføre etterbehandling samlet etter at alle oppgaver er avsluttet.
golang.org/x/sync/errgroup.ErrGroup
errgroup er en pakke som utvider sync.WaitGroup. I motsetning til sync.WaitGroup avbryter errgroup alle goroutiner og returnerer feilen hvis en feil oppstår i en av goroutinene.
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}
Koden ovenfor bruker errgroup til å opprette 10 goroutiner, der den femte goroutinen forårsaker en feil. Jeg har med vilje laget en feil i den femte goroutinen for å vise et tilfelle der feil oppstår. I praksis brukes errgroup til å opprette goroutiner, og deretter kan man utføre ulik etterbehandling for hver goroutine der det oppstår en feil.
once
Dette er et verktøy for å kjøre kode som bare skal kjøres én gang. Du kan kjøre relatert kode via konstruktørene nedenfor.
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 sikrer ganske enkelt at den aktuelle funksjonen bare kjøres é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}
Koden ovenfor bruker sync.OnceFunc til å skrive ut Hello, World!. I denne koden opprettes funksjonen once med sync.OnceFunc, og selv om once kalles flere ganger, skrives Hello, World! bare ut én gang.
OnceValue
OnceValue sikrer ikke bare at funksjonen kjøres én gang i hele programmet, men lagrer også returverdien til funksjonen, og returnerer den lagrede verdien ved nye kall.
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}
Koden ovenfor bruker sync.OnceValue til å øke variabelen c med 1. I denne koden opprettes funksjonen once med sync.OnceValue, og selv om once kalles flere ganger, økes variabelen c bare én gang, og returverdien er 1.
OnceValues
OnceValues fungerer på samme måte som OnceValue, men kan returnere flere verdier.
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}
Koden ovenfor bruker sync.OnceValues til å øke variabelen c med 1. I denne koden opprettes funksjonen once med sync.OnceValues, og selv om once kalles flere ganger, økes variabelen c bare én gang, og returverdien er 1.
atomic
Pakken atomic tilbyr atomiske operasjoner. Den tilbyr metodene Add, CompareAndSwap, Load, Store og Swap, men i nyere tid anbefales bruk av 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 eksemplet som ble brukt tidligere. Koden bruker typen atomic.Int64 til å øke variabelen c atomisk. Med metodene Add og Load kan variabelen økes atomisk og leses. I tillegg kan man lagre en verdi med metoden Store, bytte verdi med metoden Swap, og med metoden CompareAndSwap sammenligne verdien og bytte den hvis den passer.
cond
sync.Cond
Pakken cond tilbyr betingelsesvariabler. Den kan opprettes med sync.Cond og tilbyr metodene 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}
Koden ovenfor bruker sync.Cond til å vente til variabelen ready blir true. I denne koden venter sync.Cond til variabelen ready blir true, og deretter skrives Ready! ut. På denne måten kan sync.Cond la flere goroutiner vente samtidig til en bestemt betingelse er oppfylt.
Ved å bruke dette kan du implementere en enkel 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 du bruker sync.Cond på denne måten, kan du vente effektivt i stedet for å bruke mye CPU med spin-lock, og fortsette å kjøre når betingelsen er oppfylt.
semaphore
golang.org/x/sync/semaphore.Semaphore
Pakken semaphore tilbyr semaforer. Den kan opprettes med golang.org/x/sync/semaphore.Semaphore og tilbyr metodene 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}
Koden ovenfor oppretter en semafor med semaphore, henter den med metoden Acquire og frigjør den med metoden Release. I denne koden vises hvordan man henter og frigjør en semafor med semaphore.
Avslutning
Det grunnleggende innholdet burde være tilstrekkelig her. Jeg håper at du basert på innholdet i denne artikkelen kan forstå hvordan man håndterer samtidighet med goroutiner, og at du kan bruke det i praksis. Jeg håper også at denne artikkelen har vært nyttig for deg. Takk.