--- id: 01933e13f963942e4c9bc8c3c89fbc2b author: snowmerak title: Стартов пакет за конкурентност в Go description: Представяме различни инструменти и техники за управление на конкурентността в Go, включително горутини, канали и мютекси. language: bg date: 2024-10-18T05:51:30.010680581Z path: /blog/posts/go-concurrency-starter-pack-z221a3399 category: concurrency --- ## Общ преглед ### Кратко въведение Езикът Go разполага с много инструменти за управление на конкурентността. В тази статия ще представим някои от тях и няколко похвата. ### Горутини? goroutine е нов вид модел на конкурентност, поддържан от езика Go. Обикновено програмата получава от ОС нишки на ОС (OS threads), за да изпълнява няколко задачи едновременно, и изпълнява работата паралелно, колкото е броят на ядрата. За да се постигне конкурентност в по-малък мащаб, в потребителското пространство (userland) се създават зелени нишки (green threads), които се изпълняват последователно една след друга в рамките на една ОС нишка. Горутините обаче са направени още по-малки и ефективни от този вид зелени нишки. Горутините използват по-малко памет от нишките и могат да бъдат създавани и превключвани по-бързо от тях. За да използвате горутини, е достатъчно просто да използвате ключовата дума `go`. Това позволява интуитивно в процеса на писане на програмата синхронния код да се изпълнява като асинхронен. ```go package main import ( "fmt" "time" ) func main() { ch := make(chan struct{}) go func() { defer close(ch) time.Sleep(1 * time.Second) fmt.Println("Hello, World!") }() fmt.Println("Waiting for goroutine...") for range ch {} } ``` Този код преобразува синхронен код, който изчаква една секунда и след това извежда `Hello, World!`, в асинхронен поток. Текущият пример е прост, но когато по-сложен код се преобразува от синхронен в асинхронен, четливостта, видимостта и разбираемостта на кода стават по-добри, отколкото при предишни подходи като async await или promise. Въпреки това, в много случаи, без разбиране на потока при просто асинхронно извикване на такъв синхронен код и на потоци като `fork & join` (сходни с разделяй и владей), може да се получи лош код с горутини. За такива случаи в тази статия ще представим някои методи и техники. ## Управление на конкурентността ### context Може да изглежда изненадващо, че `context` е първата техника за управление. Но в Go `context` е не просто функция за отмяна, а изпълнява отлична роля при управлението на цялото дърво от задачи. Ако някои читатели не го познават, ще обясня накратко този пакет. ```go package main func main() { ctx, cancel := context.WithCancel(context.Background()) defer cancel() go func() { <-ctx.Done() fmt.Println("Context is done!") }() time.Sleep(1 * time.Second) cancel() time.Sleep(1 * time.Second) } ``` Горният код използва `context`, за да изведе `Context is done!` след една секунда. Състоянието на отмяна може да се провери чрез метода `Done()`, а методите `WithCancel`, `WithTimeout`, `WithDeadline`, `WithValue` и други предоставят различни начини за отмяна. Нека създадем прост пример. Да предположим, че пишете код, който използва шаблона `aggregator`, за да получи `user`, `post` и `comment`, и всички заявки трябва да се извършат в рамките на 2 секунди. Тогава можете да напишете следното. ```go package main func main() { ctx, cancel := context.WithTimeout(context.Background(), 2 * time.Second) defer cancel() ch := make(chan struct{}) go func() { defer close(ch) user := getUser(ctx) post := getPost(ctx) comment := getComment(ctx) fmt.Println(user, post, comment) }() select { case <-ctx.Done(): fmt.Println("Timeout!") case <-ch: fmt.Println("All data is fetched!") } } ``` Горният код извежда `Timeout!`, ако не успее да получи всички данни в рамките на 2 секунди, и извежда `All data is fetched!`, ако всички данни бъдат получени. По този начин, при използване на `context`, отмяната и таймаутът могат лесно да се управляват дори в код, в който работят множество горутини. Различни функции и методи, свързани с context, можете да намерите в [godoc context](https://pkg.go.dev/context). Надяваме се, че след като се запознаете с основите, ще можете да ги използвате удобно. ### channel #### unbuffered channel `channel` е инструмент за комуникация между горутини. `channel` може да се създаде с `make(chan T)`. Тук `T` е типът на данните, които този `channel` ще предава. Данни се изпращат и получават с `<-`, а `channel` може да се затвори с `close`. ```go package main func main() { ch := make(chan int) go func() { ch <- 1 ch <- 2 close(ch) }() for i := range ch { fmt.Println(i) } } ``` Горният код използва `channel`, за да изведе 1 и 2. В този код е показано само изпращането и получаването на стойности през `channel`. Но `channel` предлага и повече възможности. Първо ще разгледаме `buffered channel` и `unbuffered channel`. Преди да започнем, трябва да отбележим, че примерът по-горе използва `unbuffered channel`, при който действието за изпращане на данни в канала и действието за получаването им трябва да се извършат едновременно. Ако тези действия не се извършат едновременно, може да възникне deadlock. #### buffered channel Какво ще стане, ако горният код не извежда просто данни, а двата процеса извършват тежка работа? Ако вторият процес чете и обработва данни и е блокиран за дълго време, първият процес също ще спре за това време. За да предотвратим такава ситуация, можем да използваме `buffered channel`. ```go package main func main() { ch := make(chan int, 2) go func() { ch <- 1 ch <- 2 close(ch) }() for i := range ch { fmt.Println(i) } } ``` Горният код използва `buffered channel`, за да изведе 1 и 2. В този код `buffered channel` е използван така, че действията за изпращане и получаване на данни в `channel` да не трябва да се извършват едновременно. Когато каналът има буфер, той осигурява запас с дължината на буфера, поради което може да се предотврати забавяне на задачите, причинено от влиянието на последващи задачи. #### select Когато работите с няколко канала, с помощта на конструкцията `select` можете лесно да реализирате `fan-in` структура. ```go package main import ( "fmt" "time" ) func main() { ch1 := make(chan int, 10) ch2 := make(chan int, 10) ch3 := make(chan int, 10) go func() { for { ch1 <- 1 time.Sleep(1 * time.Second) } }() go func() { for { ch2 <- 2 time.Sleep(2 * time.Second) } }() go func() { for { ch3 <- 3 time.Sleep(3 * time.Second) } }() for i := 0; i < 3; i++ { select { case v := <-ch1: fmt.Println(v) case v := <-ch2: fmt.Println(v) case v := <-ch3: fmt.Println(v) } } } ``` Горният код създава 3 канала, които периодично предават 1, 2 и 3, и използва `select`, за да получава стойности от каналите и да ги извежда. По този начин, с `select`, можете едновременно да получавате данни от няколко канала и да ги обработвате веднага щом бъдат получени. #### for range `channel` може лесно да получава данни с помощта на `for range`. Когато `for range` се използва с канал, той се изпълнява при всяко добавяне на данни в канала и приключва цикъла, когато каналът бъде затворен. ```go package main func main() { ch := make(chan int) go func() { ch <- 1 ch <- 2 close(ch) }() for i := range ch { fmt.Println(i) } } ``` Горният код използва `channel`, за да изведе 1 и 2. В този код с `for range` данните се получават и извеждат при всяко добавяне в канала. Когато каналът бъде затворен, цикълът приключва. Както вече бе показано няколко пъти, този синтаксис може да се използва и като просто средство за синхронизация. ```go package main func main() { ch := make(chan struct{}) go func() { defer close(ch) time.Sleep(1 * time.Second) fmt.Println("Hello, World!") }() fmt.Println("Waiting for goroutine...") for range ch {} } ``` Горният код изчаква една секунда и след това извежда `Hello, World!`. В този код с `channel` синхронният код е преобразуван в асинхронен. По този начин, с `channel`, можете лесно да превръщате синхронен код в асинхронен и да задавате точка на `join`. #### etc 1. Ако изпратите или получите данни от nil channel, може да попаднете в безкраен цикъл и да възникне deadlock. 2. Ако изпратите данни в затворен канал, ще възникне panic. 3. Дори и да не затворите канала ръчно, GC ще го събере и затвори. ### mutex #### spinlock `spinlock` е метод за синхронизация, при който в цикъл непрекъснато се опитва да получи ключалката. В Go spinlock може лесно да се реализира с помощта на указатели. ```go package spinlock import ( "runtime" "sync/atomic" ) type SpinLock struct { lock uintptr } func (s *SpinLock) Lock() { for !atomic.CompareAndSwapUintptr(&s.lock, 0, 1) { runtime.Gosched() } } func (s *SpinLock) Unlock() { atomic.StoreUintptr(&s.lock, 0) } func NewSpinLock() *SpinLock { return &SpinLock{} } ``` Горният код е реализация на пакета `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`. ```go package main import ( "sync" ) func main() { var mu sync.Mutex var count int go func() { mu.Lock() count++ mu.Unlock() }() mu.Lock() count++ mu.Unlock() println(count) } ``` В горния код две горутини почти едновременно получават достъп до една и съща променлива `count`. Ако използвате `mutex`, за да направите кода, който достъпва променливата `count`, критична секция, можете да предотвратите едновременния достъп до `count`. Тогава този код ще извежда `2` при всяко изпълнение. #### sync.RWMutex `sync.RWMutex` е `mutex`, който позволява да се разграничават ключалки за четене и ключалки за запис. С методите `RLock` и `RUnlock` можете да поставяте и освобождавате ключалка за четене. ```go package cmap import ( "sync" ) type ConcurrentMap[K comparable, V any] struct { sync.RWMutex data map[K]V } func (m *ConcurrentMap[K, V]) Get(key K) (V, bool) { m.RLock() defer m.RUnlock() value, ok := m.data[key] return value, ok } func (m *ConcurrentMap[K, V]) Set(key K, value V) { m.Lock() defer m.Unlock() m.data[key] = value } ``` Горният код реализира `ConcurrentMap` с помощта на `sync.RWMutex`. В този код методът `Get` поставя ключалка за четене, а методът `Set` поставя ключалка за запис, така че да може безопасно да се получава достъп до картата `data` и да се променя. Ключалката за четене е необходима, защото при много прости операции за четене, вместо да се поставя ключалка за запис, се поставя само ключалка за четене, така че няколко горутини да могат да четат едновременно. По този начин, в случаите, когато не е необходимо да се променя състоянието и не е нужна ключалка за запис, производителността може да се подобри, като се постави само ключалка за четене. #### fakelock `fakelock` е прост трик за реализиране на `sync.Locker`. Тази структура предоставя същите методи като `sync.Mutex`, но всъщност не извършва никакво действие. ```go package fakelock type FakeLock struct{} func (f *FakeLock) Lock() {} func (f *FakeLock) Unlock() {} ``` Горният код е реализация на пакета `fakelock`. Този пакет реализира `sync.Locker`, като предоставя методите `Lock` и `Unlock`, но в действителност не извършва никакво действие. Защо е необходим такъв код, ще опиша при възможност по-късно. ### waitgroup #### sync.WaitGroup `sync.WaitGroup` е инструмент, който чака всички задачи на горутините да приключат. Предоставя методите `Add`, `Done` и `Wait`. С метода `Add` добавяте броя на горутините, с `Done` съобщавате, че задачата на дадена горутина е приключила, а с `Wait` чакате всички горутини да завършат задачите си. ```go package main import ( "sync" "sync/atomic" ) func main() { wg := sync.WaitGroup{} c := atomic.Int64{} for i := 0; i < 100 ; i++ { wg.Add(1) go func() { defer wg.Done() c.Add(1) }() } wg.Wait() println(c.Load()) } ``` Горният код използва `sync.WaitGroup`, за да може 100 горутини едновременно да добавят стойност към променливата `c`. В този код, с помощта на `sync.WaitGroup`, програмата изчаква всички горутини да приключат и след това извежда стойността, добавена към `c`. Когато се `fork & join` няколко задачи, е достатъчно да се използват само канали, но при `fork & join` на голям брой задачи добър избор е и `sync.WaitGroup`. #### with slice Когато се използва със слайс, `waitgroup` може да стане инструмент за управление на отлично конкурентно изпълнение на задачи без ключалки. ```go package main import ( "fmt" "sync" "rand" ) func main() { var wg sync.WaitGroup arr := [10]int{} for i := 0; i < 10; i++ { wg.Add(1) go func(id int) { defer wg.Done() arr[id] = rand.Intn(100) }(i) } wg.Wait() fmt.Println("Done") for i, v := range arr { fmt.Printf("arr[%d] = %d\n", i, v) } } ``` Горният код използва само `waitgroup`, за да може всяка горутина едновременно да генерира по едно случайно цяло число от 10 и да го запише на присвоения ѝ индекс. В този код с `waitgroup` програмата изчаква всички горутини да приключат и след това извежда `Done`. По този начин, с `waitgroup`, няколко горутини могат да работят едновременно, да записват данни без ключалки, докато всички горутини не приключат, и след това да извършат обработката след това наведнъж. #### golang.org/x/sync/errgroup.ErrGroup `errgroup` е пакет, който разширява `sync.WaitGroup`. За разлика от `sync.WaitGroup`, `errgroup`, ако при работата на някоя от горутините възникне грешка, отменя всички горутини и връща грешката. ```go package main import ( "context" "fmt" "golang.org/x/sync/errgroup" ) func main() { g, ctx := errgroup.WithContext(context.Background()) _ = ctx for i := 0; i < 10; i++ { i := i g.Go(func() error { if i == 5 { return fmt.Errorf("error") } return nil }) } if err := g.Wait(); err != nil { fmt.Println(err) } } ``` Горният код използва `errgroup`, за да създаде 10 горутини, като в петата горутина предизвиква грешка. Умишлено предизвикахме грешка в петата горутина, за да покажем случай, при който възниква грешка. Но при реално използване `errgroup` се използва за създаване на горутини, като при възникване на грешка във всяка горутина се извършва различна последваща обработка. ### once Това е инструмент за изпълнение на код, който трябва да се изпълни само веднъж. Чрез конструкторите по-долу можете да изпълните съответния код. ```go func OnceFunc(f func()) func() func OnceValue[T any](f func() T) func() T func OnceValues[T1, T2 any](f func() (T1, T2)) func() (T1, T2) ``` #### OnceFunc `OnceFunc` просто гарантира, че съответната функция ще бъде изпълнена точно веднъж за целия живот на програмата. ```go package main import "sync" func main() { once := sync.OnceFunc(func() { println("Hello, World!") }) once() once() once() once() once() } ``` Горният код използва `sync.OnceFunc`, за да извежда `Hello, World!`. В този код с `sync.OnceFunc` се създава функцията `once`, а въпреки че `once` се извиква многократно, `Hello, World!` се извежда само веднъж. #### OnceValue `OnceValue` не просто гарантира, че съответната функция се изпълнява точно веднъж, а запазва върнатата от функцията стойност и при следващи извиквания връща запазената стойност. ```go package main import "sync" func main() { c := 0 once := sync.OnceValue(func() int { c += 1 return c }) println(once()) println(once()) println(once()) println(once()) println(once()) } ``` Горният код използва `sync.OnceValue`, за да увеличава променливата `c` с 1. В този код с `sync.OnceValue` се създава функцията `once`, а въпреки че `once` се извиква многократно, променливата `c` се увеличава само веднъж и връща 1. #### OnceValues `OnceValues` работи по същия начин като `OnceValue`, но може да връща няколко стойности. ```go package main import "sync" func main() { c := 0 once := sync.OnceValues(func() (int, int) { c += 1 return c, c }) a, b := once() println(a, b) a, b = once() println(a, b) a, b = once() println(a, b) a, b = once() println(a, b) a, b = once() println(a, b) } ``` Горният код използва `sync.OnceValues`, за да увеличава променливата `c` с 1. В този код с `sync.OnceValues` се създава функцията `once`, а въпреки че `once` се извиква многократно, променливата `c` се увеличава само веднъж и връща 1. ### atomic Пакетът `atomic` предоставя атомарни операции. Той предлага методи като `Add`, `CompareAndSwap`, `Load`, `Store` и `Swap`, но напоследък се препоръчва използването на типове като `Int64`, `Uint64` и `Pointer`. ```go package main import ( "sync" "sync/atomic" ) func main() { wg := sync.WaitGroup{} c := atomic.Int64{} for i := 0; i < 100 ; i++ { wg.Add(1) go func() { defer wg.Done() c.Add(1) }() } wg.Wait() println(c.Load()) } ``` Това е пример, който използвахме преди. Кодът използва типа `atomic.Int64`, за да увеличава променливата `c` атомарно. Методите `Add` и `Load` позволяват атомарно увеличаване на променливата и четене на нейната стойност. Освен това с метода `Store` можете да запишете стойност, с `Swap` да замените стойността, а с `CompareAndSwap` да сравните стойността и, ако е подходяща, да я замените. ### cond #### sync.Cond Пакетът `cond` предоставя условни променливи. Той може да се създаде чрез `sync.Cond` и предлага методите `Wait`, `Signal` и `Broadcast`. ```go package main import ( "sync" ) func main() { c := sync.NewCond(&sync.Mutex{}) ready := false go func() { c.L.Lock() ready = true c.Signal() c.L.Unlock() }() c.L.Lock() for !ready { c.Wait() } c.L.Unlock() println("Ready!") } ``` Горният код използва `sync.Cond`, за да чака, докато променливата `ready` стане `true`. В този код с `sync.Cond` програмата чака променливата `ready` да стане `true` и след това извежда `Ready!`. По този начин, с `sync.Cond`, можете да накарате няколко горутини да чакат, докато едновременно не бъде изпълнено определено условие. Използвайки това, можете да реализирате прост `queue`. ```go package queue import ( "sync" "sync/atomic" ) type Node[T any] struct { Value T Next *Node[T] } type Queue[T any] struct { sync.Mutex Cond *sync.Cond Head *Node[T] Tail *Node[T] Len int } func New[T any]() *Queue[T] { q := &Queue[T]{} q.Cond = sync.NewCond(&q.Mutex) return q } func (q *Queue[T]) Push(value T) { q.Lock() defer q.Unlock() node := &Node[T]{Value: value} if q.Len == 0 { q.Head = node q.Tail = node } else { q.Tail.Next = node q.Tail = node } q.Len++ q.Cond.Signal() } func (q *Queue[T]) Pop() T { q.Lock() defer q.Unlock() for q.Len == 0 { q.Cond.Wait() } node := q.Head q.Head = q.Head.Next q.Len-- return node.Value } ``` Когато използвате `sync.Cond` по този начин, вместо да консумирате много процесорно време с `spin-lock`, можете ефективно да чакате и да продължите работата си, когато условието бъде изпълнено. ### semaphore #### golang.org/x/sync/semaphore.Semaphore Пакетът `semaphore` предоставя семафори. Той може да се създаде чрез `golang.org/x/sync/semaphore.Semaphore` и предлага методите `Acquire`, `Release` и `TryAcquire`. ```go package main import ( "context" "fmt" "golang.org/x/sync/semaphore" ) func main() { s := semaphore.NewWeighted(1) if s.TryAcquire(1) { fmt.Println("Acquired!") } else { fmt.Println("Not Acquired!") } s.Release(1) } ``` Горният код създава семафор с `semaphore`, придобива го с метода `Acquire` и го освобождава с метода `Release`. В този код показахме как да придобиваме и освобождаваме семафор с помощта на `semaphore`. ## Заключение Мисля, че за основното съдържание е достатъчно толкова. Надяваме се, че въз основа на тази статия ще разберете как да управлявате конкурентността с горутини и ще можете да я използвате на практика. Надяваме се, че тази статия ви е била полезна. Благодарим ви.