--- id: 01933e13f963942e4c9bc8c3c89fbc2b author: snowmerak title: Стартовый набор по конкурентности в Go description: Представляем различные инструменты и методы управления конкурентностью в Go, включая горутины, каналы, мьютексы и другие. language: ru date: 2024-10-18T05:51:30.010680581Z path: /blog/posts/go-concurrency-starter-pack-z221a3399 category: concurrency --- ## Обзор ### Краткое введение В языке Go есть множество инструментов для управления конкурентностью. В этой статье мы познакомим вас с некоторыми из них и приемами работы с ними. ### Горутины? goroutine является новой формой модели конкурентности, поддерживаемой языком Go. Обычно программа получает от ОС потоки ОС, чтобы одновременно выполнять несколько задач, и выполняет работу параллельно в количестве, равном числу ядер. Для выполнения конкурентности меньшего уровня создаются зеленые потоки в пользовательском пространстве, и несколько зеленых потоков внутри одного потока ОС поочередно выполняют работу. Однако горутины сделали этот вид зеленых потоков еще меньше и эффективнее. Горутины используют меньше памяти, чем потоки, и могут создаваться и переключаться быстрее, чем потоки. Чтобы использовать горутину, достаточно просто указать ключевое слово `go`. Это позволяет intuitively выполнять синхронный код как асинхронный в процессе написания программы. ```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 {} } ``` Этот код изменяет синхронный код, который ждет 1 секунду и выводит `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`, чтобы через 1 секунду вывести `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!") } } ``` Если все данные не получены за 2 секунды, этот код выводит `Timeout!`, а если все данные получены, выводит `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` позволяет не требовать одновременности действий отправки данных в канал и их получения. Когда в канале есть буфер, появляется запас на указанную длину, что позволяет предотвратить задержки работы, вызванные влиянием задач с более низким приоритетом. #### 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) } } } ``` Приведенный код создает три канала, которые периодически передают 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 {} } ``` Приведенный код ждет 1 секунду и выводит `Hello, World!`. В этом коде с помощью `channel` синхронный код преобразован в асинхронный. Таким образом, используя `channel`, можно легко превратить синхронный код в асинхронный и задать точку `join`. #### etc 1. Отправка или получение данных в nil channel может привести к бесконечному циклу и deadlock. 2. Отправка данных в закрытый channel вызывает panic. 3. Даже если не закрывать channel явно, 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`. Поскольку этот метод непрерывно пытается захватить блокировку, он постоянно использует CPU до получения блокировки и может попасть в бесконечный цикл. Поэтому `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) } } ``` Приведенный код создает 10 горутин с использованием `errgroup`, и в пятой горутине возникает ошибка. Ошибка намеренно вызывается в пятой горутине, чтобы показать случай возникновения ошибки. Однако на практике `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` вместо большого расхода CPU на `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`. ## Заключение Пожалуй, для базового материала достаточно этого. Надеюсь, на основе этой статьи вы поймете, как управлять конкурентностью с помощью горутин, и сможете применять это на практике. Надеюсь, эта статья была вам полезна. Спасибо.