GoSudaСтартов пакет за конкурентност в Go
Общ преглед
Кратко въведение
Езикът Go разполага с много инструменти за управление на конкурентността. В тази статия ще представим някои от тях и няколко похвата.
Горутини?
goroutine е нов вид модел на конкурентност, поддържан от езика Go. Обикновено програмата получава от ОС нишки на ОС (OS threads), за да изпълнява няколко задачи едновременно, и изпълнява работата паралелно, колкото е броят на ядрата. За да се постигне конкурентност в по-малък мащаб, в потребителското пространство (userland) се създават зелени нишки (green threads), които се изпълняват последователно една след друга в рамките на една ОС нишка. Горутините обаче са направени още по-малки и ефективни от този вид зелени нишки. Горутините използват по-малко памет от нишките и могат да бъдат създавани и превключвани по-бързо от тях.
За да използвате горутини, е достатъчно просто да използвате ключовата дума go. Това позволява интуитивно в процеса на писане на програмата синхронния код да се изпълнява като асинхронен.
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}
Този код преобразува синхронен код, който изчаква една секунда и след това извежда Hello, World!, в асинхронен поток. Текущият пример е прост, но когато по-сложен код се преобразува от синхронен в асинхронен, четливостта, видимостта и разбираемостта на кода стават по-добри, отколкото при предишни подходи като async await или promise.
Въпреки това, в много случаи, без разбиране на потока при просто асинхронно извикване на такъв синхронен код и на потоци като fork & join (сходни с разделяй и владей), може да се получи лош код с горутини. За такива случаи в тази статия ще представим някои методи и техники.
Управление на конкурентността
context
Може да изглежда изненадващо, че context е първата техника за управление. Но в Go context е не просто функция за отмяна, а изпълнява отлична роля при управлението на цялото дърво от задачи. Ако някои читатели не го познават, ще обясня накратко този пакет.
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}
Горният код използва context, за да изведе Context is done! след една секунда. Състоянието на отмяна може да се провери чрез метода Done(), а методите WithCancel, WithTimeout, WithDeadline, WithValue и други предоставят различни начини за отмяна.
Нека създадем прост пример. Да предположим, че пишете код, който използва шаблона aggregator, за да получи user, post и comment, и всички заявки трябва да се извършат в рамките на 2 секунди. Тогава можете да напишете следното.
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}
Горният код извежда Timeout!, ако не успее да получи всички данни в рамките на 2 секунди, и извежда All data is fetched!, ако всички данни бъдат получени. По този начин, при използване на context, отмяната и таймаутът могат лесно да се управляват дори в код, в който работят множество горутини.
Различни функции и методи, свързани с context, можете да намерите в godoc context. Надяваме се, че след като се запознаете с основите, ще можете да ги използвате удобно.
channel
unbuffered channel
channel е инструмент за комуникация между горутини. channel може да се създаде с make(chan T). Тук T е типът на данните, които този channel ще предава. Данни се изпращат и получават с <-, а channel може да се затвори с 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}
Горният код използва channel, за да изведе 1 и 2. В този код е показано само изпращането и получаването на стойности през channel. Но channel предлага и повече възможности. Първо ще разгледаме buffered channel и unbuffered channel. Преди да започнем, трябва да отбележим, че примерът по-горе използва unbuffered channel, при който действието за изпращане на данни в канала и действието за получаването им трябва да се извършат едновременно. Ако тези действия не се извършат едновременно, може да възникне deadlock.
buffered channel
Какво ще стане, ако горният код не извежда просто данни, а двата процеса извършват тежка работа? Ако вторият процес чете и обработва данни и е блокиран за дълго време, първият процес също ще спре за това време. За да предотвратим такава ситуация, можем да използваме 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}
Горният код използва buffered channel, за да изведе 1 и 2. В този код buffered channel е използван така, че действията за изпращане и получаване на данни в channel да не трябва да се извършват едновременно. Когато каналът има буфер, той осигурява запас с дължината на буфера, поради което може да се предотврати забавяне на задачите, причинено от влиянието на последващи задачи.
select
Когато работите с няколко канала, с помощта на конструкцията select можете лесно да реализирате fan-in структура.
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}
Горният код създава 3 канала, които периодично предават 1, 2 и 3, и използва select, за да получава стойности от каналите и да ги извежда. По този начин, с select, можете едновременно да получавате данни от няколко канала и да ги обработвате веднага щом бъдат получени.
for range
channel може лесно да получава данни с помощта на for range. Когато for range се използва с канал, той се изпълнява при всяко добавяне на данни в канала и приключва цикъла, когато каналът бъде затворен.
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}
Горният код използва channel, за да изведе 1 и 2. В този код с for range данните се получават и извеждат при всяко добавяне в канала. Когато каналът бъде затворен, цикълът приключва.
Както вече бе показано няколко пъти, този синтаксис може да се използва и като просто средство за синхронизация.
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}
Горният код изчаква една секунда и след това извежда Hello, World!. В този код с channel синхронният код е преобразуван в асинхронен. По този начин, с channel, можете лесно да превръщате синхронен код в асинхронен и да задавате точка на join.
etc
- Ако изпратите или получите данни от nil channel, може да попаднете в безкраен цикъл и да възникне deadlock.
- Ако изпратите данни в затворен канал, ще възникне panic.
- Дори и да не затворите канала ръчно, GC ще го събере и затвори.
mutex
spinlock
spinlock е метод за синхронизация, при който в цикъл непрекъснато се опитва да получи ключалката. В Go spinlock може лесно да се реализира с помощта на указатели.
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}
Горният код е реализация на пакета spinlock. В този код SpinLock е реализиран с помощта на пакета sync/atomic. Методът Lock използва atomic.CompareAndSwapUintptr, за да опита да получи ключалката, а методът Unlock използва atomic.StoreUintptr, за да я освободи. Тъй като този метод непрекъснато опитва да получи ключалката, докато не я получи, той през цялото време използва процесора и може да попадне в безкраен цикъл. Затова е препоръчително spinlock да се използва за проста синхронизация или само за кратки интервали от време.
sync.Mutex
mutex е инструмент за синхронизация между горутини. mutex, предоставен от пакета sync, предлага методи като Lock, Unlock, RLock и RUnlock. mutex може да се създаде с sync.Mutex, а за заключване при четене/запис може да се използва 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}
В горния код две горутини почти едновременно получават достъп до една и съща променлива count. Ако използвате mutex, за да направите кода, който достъпва променливата count, критична секция, можете да предотвратите едновременния достъп до count. Тогава този код ще извежда 2 при всяко изпълнение.
sync.RWMutex
sync.RWMutex е mutex, който позволява да се разграничават ключалки за четене и ключалки за запис. С методите RLock и 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}
Горният код реализира ConcurrentMap с помощта на sync.RWMutex. В този код методът Get поставя ключалка за четене, а методът Set поставя ключалка за запис, така че да може безопасно да се получава достъп до картата data и да се променя. Ключалката за четене е необходима, защото при много прости операции за четене, вместо да се поставя ключалка за запис, се поставя само ключалка за четене, така че няколко горутини да могат да четат едновременно. По този начин, в случаите, когато не е необходимо да се променя състоянието и не е нужна ключалка за запис, производителността може да се подобри, като се постави само ключалка за четене.
fakelock
fakelock е прост трик за реализиране на sync.Locker. Тази структура предоставя същите методи като sync.Mutex, но всъщност не извършва никакво действие.
1package fakelock
2
3type FakeLock struct{}
4
5func (f *FakeLock) Lock() {}
6
7func (f *FakeLock) Unlock() {}
Горният код е реализация на пакета fakelock. Този пакет реализира sync.Locker, като предоставя методите Lock и Unlock, но в действителност не извършва никакво действие. Защо е необходим такъв код, ще опиша при възможност по-късно.
waitgroup
sync.WaitGroup
sync.WaitGroup е инструмент, който чака всички задачи на горутините да приключат. Предоставя методите Add, Done и Wait. С метода Add добавяте броя на горутините, с Done съобщавате, че задачата на дадена горутина е приключила, а с Wait чакате всички горутини да завършат задачите си.
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}
Горният код използва sync.WaitGroup, за да може 100 горутини едновременно да добавят стойност към променливата c. В този код, с помощта на sync.WaitGroup, програмата изчаква всички горутини да приключат и след това извежда стойността, добавена към c. Когато се fork & join няколко задачи, е достатъчно да се използват само канали, но при fork & join на голям брой задачи добър избор е и sync.WaitGroup.
with slice
Когато се използва със слайс, waitgroup може да стане инструмент за управление на отлично конкурентно изпълнение на задачи без ключалки.
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}
Горният код използва само waitgroup, за да може всяка горутина едновременно да генерира по едно случайно цяло число от 10 и да го запише на присвоения ѝ индекс. В този код с waitgroup програмата изчаква всички горутини да приключат и след това извежда Done. По този начин, с waitgroup, няколко горутини могат да работят едновременно, да записват данни без ключалки, докато всички горутини не приключат, и след това да извършат обработката след това наведнъж.
golang.org/x/sync/errgroup.ErrGroup
errgroup е пакет, който разширява sync.WaitGroup. За разлика от sync.WaitGroup, errgroup, ако при работата на някоя от горутините възникне грешка, отменя всички горутини и връща грешката.
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}
Горният код използва errgroup, за да създаде 10 горутини, като в петата горутина предизвиква грешка. Умишлено предизвикахме грешка в петата горутина, за да покажем случай, при който възниква грешка. Но при реално използване errgroup се използва за създаване на горутини, като при възникване на грешка във всяка горутина се извършва различна последваща обработка.
once
Това е инструмент за изпълнение на код, който трябва да се изпълни само веднъж. Чрез конструкторите по-долу можете да изпълните съответния код.
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 просто гарантира, че съответната функция ще бъде изпълнена точно веднъж за целия живот на програмата.
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}
Горният код използва sync.OnceFunc, за да извежда Hello, World!. В този код с sync.OnceFunc се създава функцията once, а въпреки че once се извиква многократно, Hello, World! се извежда само веднъж.
OnceValue
OnceValue не просто гарантира, че съответната функция се изпълнява точно веднъж, а запазва върнатата от функцията стойност и при следващи извиквания връща запазената стойност.
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}
Горният код използва sync.OnceValue, за да увеличава променливата c с 1. В този код с sync.OnceValue се създава функцията once, а въпреки че once се извиква многократно, променливата c се увеличава само веднъж и връща 1.
OnceValues
OnceValues работи по същия начин като OnceValue, но може да връща няколко стойности.
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}
Горният код използва sync.OnceValues, за да увеличава променливата c с 1. В този код с sync.OnceValues се създава функцията once, а въпреки че once се извиква многократно, променливата c се увеличава само веднъж и връща 1.
atomic
Пакетът atomic предоставя атомарни операции. Той предлага методи като Add, CompareAndSwap, Load, Store и Swap, но напоследък се препоръчва използването на типове като 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}
Това е пример, който използвахме преди. Кодът използва типа atomic.Int64, за да увеличава променливата c атомарно. Методите Add и Load позволяват атомарно увеличаване на променливата и четене на нейната стойност. Освен това с метода Store можете да запишете стойност, с Swap да замените стойността, а с CompareAndSwap да сравните стойността и, ако е подходяща, да я замените.
cond
sync.Cond
Пакетът cond предоставя условни променливи. Той може да се създаде чрез sync.Cond и предлага методите 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}
Горният код използва sync.Cond, за да чака, докато променливата ready стане true. В този код с sync.Cond програмата чака променливата ready да стане true и след това извежда Ready!. По този начин, с sync.Cond, можете да накарате няколко горутини да чакат, докато едновременно не бъде изпълнено определено условие.
Използвайки това, можете да реализирате прост 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}
Когато използвате sync.Cond по този начин, вместо да консумирате много процесорно време с spin-lock, можете ефективно да чакате и да продължите работата си, когато условието бъде изпълнено.
semaphore
golang.org/x/sync/semaphore.Semaphore
Пакетът semaphore предоставя семафори. Той може да се създаде чрез golang.org/x/sync/semaphore.Semaphore и предлага методите 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}
Горният код създава семафор с semaphore, придобива го с метода Acquire и го освобождава с метода Release. В този код показахме как да придобиваме и освобождаваме семафор с помощта на semaphore.
Заключение
Мисля, че за основното съдържание е достатъчно толкова. Надяваме се, че въз основа на тази статия ще разберете как да управлявате конкурентността с горутини и ще можете да я използвате на практика. Надяваме се, че тази статия ви е била полезна. Благодарим ви.