GoSudaGo-startpaket för samtidighet
Översikt
Kort introduktion
Go-språket har många verktyg för hantering av samtidighet. I den här artikeln vill vi presentera några av dem samt några knep.
Goroutiner?
En goroutine är en ny form av samtidighetsmodell som stöds av Go-språket. Normalt får ett program OS-trådar från operativsystemet för att utföra flera uppgifter samtidigt, och utför arbete parallellt så många gånger som antalet kärnor. För att utföra samtidighet i mindre skala skapar man gröna trådar i användarutrymmet, så att flera gröna trådar kör arbete i omgångar inom en enda OS-tråd. Goroutiner har dock gjorts ännu mindre och mer effektiva än denna form av gröna trådar. Goroutiner använder mindre minne än trådar och kan skapas och bytas ut snabbare än trådar.
För att använda en goroutine behöver du bara använda nyckelordet go. Detta gör det intuitivt möjligt att under programmeringsprocessen köra synkron kod som asynkron kod.
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}
Denna kod ändrar en synkron kod, som vilar en sekund och sedan skriver ut Hello, World!, till ett asynkront flöde. Även om exemplet nu är enkelt, blir läsbarheten, synligheten och förståelsen av koden bättre än med tidigare metoder som async await eller promise, när man omvandlar något mer komplicerad kod från synkron till asynkron kod.
I många fall kan dock dåliga goroutinkoder uppstå när man inte förstår flödet att helt enkelt anropa synkron kod asynkront, eller flöden som fork & join (ett flöde som liknar divide and conquer). För att förbereda sig för sådana fall kommer den här artikeln att presentera några metoder och tekniker.
Hantering av samtidighet
context
Att context förekommer som första hanteringsteknik kan vara överraskande. Men i Go spelar context en utmärkt roll för att hantera hela arbetsträdet, inte bara för enkel avbrytning. För dem som inte känner till paketet ska vi kort förklara det.
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}
Ovanstående kod använder context och skriver ut Context is done! efter en sekund. context kan kontrollera om avbrytning har skett via metoden Done(), och erbjuder olika sätt att avbryta genom metoder som WithCancel, WithTimeout, WithDeadline och WithValue.
Låt oss skapa ett enkelt exempel. Antag att du vill skriva kod som hämtar user, post och comment med ett aggregator-mönster för att hämta data, och att alla förfrågningar måste slutföras inom två sekunder. Då kan du skriva följande.
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}
Ovanstående kod skriver ut Timeout! om alla data inte hämtas inom två sekunder, och skriver ut All data is fetched! om alla data hämtas. På detta sätt kan man med context enkelt hantera avbrytning och timeout även i kod där flera goroutiner körs.
Olika funktioner och metoder relaterade till context finns i godoc context. Vi hoppas att du kan lära dig de enkla delarna och använda dem bekvämt.
channel
unbuffered channel
channel är ett verktyg för kommunikation mellan goroutiner. En channel skapas med make(chan T). Här är T den typ av data som kanalen ska överföra. Data skickas och tas emot med <-, och en channel stängs 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}
Ovanstående kod använder channel för att skriva ut 1 och 2. Denna kod visar endast enkel sändning och mottagning av värden till en kanal. Men channel erbjuder fler funktioner än så. Låt oss först titta på buffered channel och unbuffered channel. Innan vi börjar bör vi notera att exemplet ovan använder en unbuffered channel, där sändningen av data till kanalen och mottagningen av data måste ske samtidigt. Om dessa åtgärder inte sker samtidigt kan en deadlock uppstå.
buffered channel
Vad händer om ovanstående kod, i stället för en enkel utskrift, är två processer som utför tunga uppgifter? Om den andra processen tar emot och bearbetar data men fastnar under lång tid, kommer även den första processen att stoppas under den tiden. För att förhindra en sådan situation kan vi använda en 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}
Ovanstående kod använder buffered channel för att skriva ut 1 och 2. I denna kod används en buffered channel så att sändningen och mottagningen av data till kanalen inte behöver ske samtidigt. När kanalen har en buffert finns det utrymme motsvarande dess längd, vilket förhindrar fördröjningar i arbetet orsakade av påverkan från efterföljande uppgifter.
select
När man hanterar flera kanaler kan man enkelt implementera en fan-in-struktur med hjälp av select-syntaxen.
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}
Ovanstående kod skapar tre kanaler som periodiskt skickar 1, 2 och 3, och använder select för att ta emot värden från kanalerna och skriva ut dem. På detta sätt kan man med select ta emot data från flera kanaler samtidigt och bearbeta värdena i den ordning de tas emot.
for range
En channel kan enkelt ta emot data med for range. När for range används på en kanal körs loopen varje gång data läggs till i kanalen, och loopen avslutas när kanalen stängs.
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}
Ovanstående kod använder channel för att skriva ut 1 och 2. I denna kod används for range för att ta emot och skriva ut data varje gång data läggs till i kanalen. När kanalen stängs avslutas loopen.
Som vi skrivit några gånger ovan kan denna syntax också användas som ett enkelt synkroniseringsverktyg.
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}
Ovanstående kod vilar en sekund och skriver sedan ut Hello, World!. I denna kod används channel för att omvandla synkron kod till asynkron kod. På detta sätt kan man med channel enkelt omvandla synkron kod till asynkron kod och även definiera en join-punkt.
etc
- Om man skickar eller tar emot data på en nil channel kan programmet fastna i en oändlig loop, vilket leder till en deadlock.
- Om man skickar data efter att en kanal har stängts uppstår en panic.
- Även om man inte stänger en kanal explicit, kommer GC att samla in den och stänga kanalen.
mutex
spinlock
spinlock är en synkroniseringsmetod som upprepade gånger försöker ta låset i en loop. I Go kan en spinlock enkelt implementeras med hjälp av pekare.
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}
Ovanstående kod implementerar paketet spinlock. Här implementeras SpinLock med hjälp av paketet sync/atomic. I metoden Lock används atomic.CompareAndSwapUintptr för att försöka ta låset, och i metoden Unlock används atomic.StoreUintptr för att frigöra låset. Eftersom denna metod försöker ta låset utan paus, använder den CPU kontinuerligt tills låset erhålls, vilket kan leda till en oändlig loop. Därför är spinlock lämplig att använda för enkel synkronisering eller endast under kort tid.
sync.Mutex
mutex är ett verktyg för synkronisering mellan goroutiner. mutex som tillhandahålls av paketet sync erbjuder metoderna Lock, Unlock, RLock, RUnlock med flera. En mutex skapas med sync.Mutex, och sync.RWMutex kan användas för läs- och skrivlås.
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 ovanstående kod får två goroutiner nästan samtidigt åtkomst till samma variabel count. Om man använder mutex för att göra koden som kommer åt variabeln count till ett kritiskt avsnitt, kan samtidig åtkomst till count förhindras. Då skriver denna kod ut 2 varje gång den körs.
sync.RWMutex
sync.RWMutex är en mutex som kan användas genom att skilja mellan läslås och skrivlås. Med metoderna RLock och RUnlock kan ett läslås sättas och släppas.
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}
Ovanstående kod implementerar ConcurrentMap med hjälp av sync.RWMutex. I denna kod sätts ett läslås i metoden Get och ett skrivlås i metoden Set, vilket gör det möjligt att säkert komma åt och ändra mappen data. Anledningen till att läslås behövs är att när det finns många enkla läsoperationer, kan flera goroutiner utföra läsoperationer samtidigt genom att endast sätta ett läslås, utan att sätta ett skrivlås. På så sätt kan prestandan förbättras i de fall där det inte sker någon förändring av tillståndet och därmed inte är nödvändigt att sätta ett skrivlås.
fakelock
fakelock är ett enkelt knep som implementerar sync.Locker. Denna struktur tillhandahåller samma metoder som sync.Mutex, men utför faktiskt ingen funktion.
1package fakelock
2
3type FakeLock struct{}
4
5func (f *FakeLock) Lock() {}
6
7func (f *FakeLock) Unlock() {}
Ovanstående kod implementerar paketet fakelock. Paketet implementerar sync.Locker och tillhandahåller metoderna Lock och Unlock, men gör i praktiken ingenting. Varför en sådan kod behövs kommer jag att beskriva vid tillfälle.
waitgroup
sync.WaitGroup
sync.WaitGroup är ett verktyg som väntar tills alla goroutinernas uppgifter är klara. Det tillhandahåller metoderna Add, Done och Wait. Med metoden Add lägger man till antalet goroutiner, med metoden Done meddelar man att en goroutins uppgift är klar, och med metoden Wait väntar man tills alla goroutiners uppgifter är klara.
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}
Ovanstående kod använder sync.WaitGroup så att 100 goroutiner samtidigt adderar ett värde till variabeln c. I denna kod väntar sync.WaitGroup tills alla goroutiner är klara, och skriver sedan ut det värde som adderats till c. När man bara fork & join:ar ett fåtal uppgifter räcker det med kanaler, men när man fork & join:ar ett stort antal uppgifter är sync.WaitGroup ett bra alternativ.
with slice
När den används tillsammans med slice kan waitgroup bli ett utmärkt verktyg för att hantera samtidig exekvering utan 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}
Ovanstående kod använder endast waitgroup så att varje goroutin samtidigt genererar 10 slumpmässiga heltal och lagrar dem på det tilldelade indexet. I denna kod väntar waitgroup tills alla goroutiner är klara, och skriver sedan ut Done. På detta sätt kan man med waitgroup låta flera goroutiner utföra arbete samtidigt, lagra data utan lås tills alla goroutiner är klara, och sedan utföra efterbearbetning i ett svep efter att arbetet avslutats.
golang.org/x/sync/errgroup.ErrGroup
errgroup är ett paket som utökar sync.WaitGroup. Till skillnad från sync.WaitGroup avbryter errgroup alla goroutiner och returnerar felet om ett fel uppstår i någon av goroutinernas uppgifter.
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}
Ovanstående kod använder errgroup för att skapa tio goroutiner, och genererar ett fel i den femte goroutinen. Eftersom felet avsiktligt genereras i den femte goroutinen visas ett exempel på en felsituation. I praktiken används errgroup genom att skapa goroutiner med det, och sedan tillämpa olika efterbearbetningar när fel uppstår i respektive goroutin.
once
Det är ett verktyg för att köra kod som endast ska köras en gång. Relaterad kod kan köras via konstruktorerna nedan.
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 garanterar helt enkelt att den givna funktionen endast kan köras exakt en gång under hela programmets livstid.
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}
Ovanstående kod använder sync.OnceFunc för att skriva ut Hello, World!. I denna kod används sync.OnceFunc för att skapa funktionen once, och även om once anropas flera gånger skrivs Hello, World! endast ut en gång.
OnceValue
OnceValue körs inte bara en gång under hela programmets livstid; den sparar dessutom funktionens returvärde och returnerar det sparade värdet när den anropas igen.
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}
Ovanstående kod använder sync.OnceValue för att öka variabeln c med 1. I denna kod används sync.OnceValue för att skapa funktionen once, och även om once anropas flera gånger returnerar den 1, eftersom variabeln c endast ökades en gång.
OnceValues
OnceValues fungerar på samma sätt som OnceValue, men kan returnera flera värden.
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}
Ovanstående kod använder sync.OnceValues för att öka variabeln c med 1. I denna kod används sync.OnceValues för att skapa funktionen once, och även om once anropas flera gånger returnerar den 1, eftersom variabeln c endast ökades en gång.
atomic
Paketet atomic är ett paket som tillhandahåller atomära operationer. Paketet atomic erbjuder metoder som Add, CompareAndSwap, Load, Store och Swap, men på senare tid rekommenderas användning av typer som Int64, Uint64 och 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}
Detta är exemplet som användes tidigare. Det använder typen atomic.Int64 för att öka variabeln c atomärt. Med metoderna Add och Load kan variabeln ökas atomärt och läsas. Dessutom kan man med metoden Store lagra ett värde, med metoden Swap byta ut ett värde, och med metoden CompareAndSwap jämföra ett värde och byta ut det om det är lämpligt.
cond
sync.Cond
Paketet cond är ett paket som tillhandahåller villkorsvariabler. Paketet cond kan skapas med sync.Cond och erbjuder metoderna Wait, Signal och 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}
Ovanstående kod använder sync.Cond för att vänta tills variabeln ready blir true. I denna kod används sync.Cond för att vänta tills variabeln ready blir true, och därefter skrivs Ready! ut. På detta sätt kan man med sync.Cond låta flera goroutiner vänta tills ett specifikt villkor är uppfyllt.
Med detta kan man implementera 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 man använder sync.Cond på detta sätt kan man vänta effektivt och återuppta arbetet när villkoret är uppfyllt, i stället för att använda mycket CPU med spin-lock.
semaphore
golang.org/x/sync/semaphore.Semaphore
Paketet semaphore är ett paket som tillhandahåller semaforer. Paketet semaphore kan skapas med golang.org/x/sync/semaphore.Semaphore och erbjuder metoderna Acquire, Release och 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}
Ovanstående kod skapar en semafor med semaphore, och använder metoden Acquire för att ta semaforen och metoden Release för att frigöra den. I denna kod visas hur man tar och frigör en semafor med semaphore.
Avslutning
Det grundläggande innehållet tror vi räcker här. Utifrån innehållet i den här artikeln hoppas vi att du förstår hur man hanterar samtidighet med goroutiner och kan använda det i praktiken. Vi hoppas att den här artikeln har varit till nytta för dig. Tack.