GoSudaСтартовый набор по конкурентности в Go
Обзор
Краткое введение
В языке Go есть множество инструментов для управления конкурентностью. В этой статье мы познакомим вас с некоторыми из них и приемами работы с ними.
Горутины?
goroutine является новой формой модели конкурентности, поддерживаемой языком Go. Обычно программа получает от ОС потоки ОС, чтобы одновременно выполнять несколько задач, и выполняет работу параллельно в количестве, равном числу ядер. Для выполнения конкурентности меньшего уровня создаются зеленые потоки в пользовательском пространстве, и несколько зеленых потоков внутри одного потока ОС поочередно выполняют работу. Однако горутины сделали этот вид зеленых потоков еще меньше и эффективнее. Горутины используют меньше памяти, чем потоки, и могут создаваться и переключаться быстрее, чем потоки.
Чтобы использовать горутину, достаточно просто указать ключевое слово go. Это позволяет intuitively выполнять синхронный код как асинхронный в процессе написания программы.
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}
Этот код изменяет синхронный код, который ждет 1 секунду и выводит 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, чтобы через 1 секунду вывести 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}
Если все данные не получены за 2 секунды, этот код выводит Timeout!, а если все данные получены, выводит 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 позволяет не требовать одновременности действий отправки данных в канал и их получения. Когда в канале есть буфер, появляется запас на указанную длину, что позволяет предотвратить задержки работы, вызванные влиянием задач с более низким приоритетом.
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}
Приведенный код создает три канала, которые периодически передают 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}
Приведенный код ждет 1 секунду и выводит Hello, World!. В этом коде с помощью channel синхронный код преобразован в асинхронный. Таким образом, используя channel, можно легко превратить синхронный код в асинхронный и задать точку join.
etc
- Отправка или получение данных в nil channel может привести к бесконечному циклу и deadlock.
- Отправка данных в закрытый channel вызывает panic.
- Даже если не закрывать channel явно, 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. Поскольку этот метод непрерывно пытается захватить блокировку, он постоянно использует CPU до получения блокировки и может попасть в бесконечный цикл. Поэтому 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}
Приведенный код создает 10 горутин с использованием errgroup, и в пятой горутине возникает ошибка. Ошибка намеренно вызывается в пятой горутине, чтобы показать случай возникновения ошибки. Однако на практике 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 вместо большого расхода CPU на 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.
Заключение
Пожалуй, для базового материала достаточно этого. Надеюсь, на основе этой статьи вы поймете, как управлять конкурентностью с помощью горутин, и сможете применять это на практике. Надеюсь, эта статья была вам полезна. Спасибо.