Разбираем паттерны конкурентности

База

Параллельность - выполнение задач в один момент времени на разных логических ядрах.
Конкурентность - выполнение задач последовательно, но со сменой контекста на другую задачу в ожидание завершения иной задачи. У пользователя может возникнуть иллюзия многозадачности даже в однопроцессорной системе, поскольку смена контекста происходит быстро (микросекунды).

Процессы:

• Раздельная память

• Раздельные ресурсы

• Раздельные регистры

Потоки:

• Общая память

• Общие ресурсы

• Раздельные стэк и регистры

Горутины:

• Общая память

• Общие ресурсы

• Общий системный стэк

• Общие регистры

Go runtime представляет модель P:M:G.
P - представляет логическое ядро процессора.
M - поток ОС по числу процессоров P.
G - структура, которая выполняет переданную функцию, создаётся по необходимости, минимум одна на старте программы (main). Стэк всего 2кб, может расширятся до 1гб для 64x и до 250кб для 32х систем.

Управление горутинами осуществляется планировщиком Go, а не ОС. Планировщик Go работает в пользовательском пространстве. Мы не можем напрямую управлять на каком процессоре будет исполняться горутина, за это отвечает планировщик.

Канал - очередь сообщений, которая умеет работать в многопоточной среде, работает по принципу FIFO.
Есть два типа каналов: буферизованный и небуферизованный.
Первый может хранить несколько сообщений, второй только одно.

Синхронизация

sync.WaitGroup - счётчик, который позволяет подождать завершения горутин.
sync.Mutex - блокирует доступ к ресурсу.
sync.RwMutex - разделяемая блокировка на чтение и запись. Читать могут несколько горутин, но мутировать данные только одна.
sync.Atomic - атомарная операция чтения и записи. Работает только с простыми значениями.
sync.Map - lock-free структура. Работает так же, как и обычная map, но потокобезопасная, можно использовать в многопоточной среде. Хорошо подходит для случаев, где надо много читать и мало писать. Если надо много писать, то лучше использовать обычную map и sync.RwMutex.

Небуферизованный канал

Буферизованный канал

Ограничения канала

Важные правила

• Закрывает канал тот, кто в него пишет.

• Если пишет несколько продюсеров, то закрывает тот, кто создал продюсеров.

• Не закрытый канал держит ресурсы. Закрывать надо явно.

Паттерны

Generator - микропаттерн, который наполняет канал. Закрываем канал, чтобы не было проблем.

func generator() <- chan int {
	ch := make(chant int)
	go func(){
		for i := 0; i <= 12; i++ {
			ch <- i + 1
		}
		close(ch)
	}()
	return ch
}

Wrapper - оборачиваем функцию, добавляя функциональность. Если вам что-то говорит слово декоратор, то это тот самый паттерн.

func wrapper(wg *sync.WaitGroup, fn func()) {  
    wg.Add(1)  
  
    go func() {  
       defer wg.Done()  
  
       fmt.Println("Work before func")  
       fn()  
       time.Sleep(1 * time.Second)  
       fmt.Println("Work after func")  
    }()  
}  
  
func main() {  
    var wg sync.WaitGroup  
  
    wrapper(&wg, func() {  
       time.Sleep(1 * time.Second)  
       fmt.Println("heavy work")  
    })  
  
    wg.Wait()  
}

Fan-in - собирает результаты из нескольких каналов в один.

func fanIn(input1, input2 <-chan string) <-chan string {
	ch := make(chan string)
	go func(){
		for {
			select {
				case s := <-input1: ch <- s
				case s := <-input2: ch <- s
			}
		}
	}()
	return ch
}

Fan-out - одна или несколько горутин пишут в канал, с другой стороны рабочие горутины читают канал, делают работу и умирают.

func worker(ch <-chan int, wg *sync.WaitGroup) {  
    wg.Done()  
    for v := range ch {  
       fmt.Println(v)  
       time.Sleep(1 * time.Second)  
    }  
}  
  
func sender() {  
    ch := make(chan int)  
    var wg sync.WaitGroup  
    for i := 0; i < 2; i++ {  
       wg.Add(1)  
       go worker(ch, &wg)  
    }  
  
    for i := 0; i < 10; i++ {  
       ch <- i  
    }  
  
    close(ch)  
  
    wg.Wait()  
    fmt.Println("done")  
}

Pipeline - данные обрабатываются цепочкой. Producer -> Producer/Consumer -> Consumer. Стадий обработки может быть сколько угодно.

func producer() <-chan int {  
    c := make(chan int)  
  
    go func() {  
       for i := 0; i <= 10; i++ {  
          c <- i + 1  
       }  
       close(c)  
    }()  
  
    return c  
}  
  
func producerConsumer(c <-chan int) <-chan int {  
    out := make(chan int)  
  
    go func() {  
       for v := range c {  
          out <- v * 2  
       }  
       close(out)  
    }()  
  
    return out  
}  
  
func consumer(ch <-chan int) {  
    for v := range ch {  
       fmt.Println(v)  
    }  
}

Cancellation - Способ прерывания горутин. Необходим, чтобы избегать висящих горутин, останавливать слишком долгие операции.

// 1. WithCancel
func worker(ctx context.Context) {  
    for {  
       select {  
       case <-ctx.Done():  
          fmt.Println("Done")  
          return  
       default:  
          fmt.Println("Working...")  
          time.Sleep(500 * time.Millisecond)  
       }  
    }  
}  
  
func main() {  
    ctx, cancel := context.WithCancel(context.Background())  
    go worker(ctx)  
  
    time.Sleep(1 * time.Second) // работаем  
    cancel()  // отменяем
    time.Sleep(1 * time.Second)  // время на завершение
}

// 2. WithTimeout
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) // спустя 2 секунды воркер перестанет работать 

// 3. WithDeadline. Можно указать точное время остановки.
ctx, cancel := context.WithDeadline(context.Background(), time.Now().Add(2*time.Second)) // прибавляем к текущему времени две секунды

Worker pool - каждый воркер берёт задачу, делает работу и отправляет результат в канал, другая горутина, в нашем случае main, читает результат из канала.

func worker(jobs <-chan int, results chan<- int, wg *sync.WaitGroup) {  
    defer wg.Done()  
  
    for j := range jobs {  
       time.Sleep(1 * time.Second)  
       fmt.Println("job", j)  
       results <- j * j  
    }  
  
}  
  
func main() {  
    jobs := make(chan int)  
    results := make(chan int)  
  
    var wg sync.WaitGroup  
  
    for i := 0; i < 3; i++ {  
       wg.Add(1)  
       go worker(jobs, results, &wg)  
    }  
  
    go func() {  
       for i := 0; i < 10; i++ {  
          jobs <- i  
       }  
       close(jobs)  
    }()  
  
    go func() {  
       wg.Wait()  
       close(results)  
    }()  
  
    for result := range results {  
       fmt.Println(result)  
    }  
}