База
Параллельность - выполнение задач в один момент времени на разных логических ядрах.
Конкурентность - выполнение задач последовательно, но со сменой контекста на другую задачу в ожидание завершения иной задачи. У пользователя может возникнуть иллюзия многозадачности даже в однопроцессорной системе, поскольку смена контекста происходит быстро (микросекунды).
Процессы:
Раздельная память
Раздельные ресурсы
Раздельные регистры
Потоки:
Общая память
Общие ресурсы
Раздельные стэк и регистры
Горутины:
Общая память
Общие ресурсы
Общий системный стэк
Общие регистры
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.
Небуферизованный канал
Действие | Открытый канал | Закрытый канал | Неинициализированный канал |
|---|---|---|---|
Чтение | Блокировка до прихода писателя | Zero value | Блокировка навсегда |
Запись | Блокировка до прихода читателя | Panic | Блокировка навсегда |
Закрытие | Канал закроется | Panic | Panic |
Буферизованный канал
Действие | Открытый и частично заполненный | Открытый и полностью заполненный | Открытый и пустой | Закрытый и частично заполненный |
|---|---|---|---|---|
Чтение | Прочитаем значение | Прочитаем значение | Блокировка до прихода писателя | Прочитаем значение |
Запись | Запишем значение | Блокировка до прихода читателя | Запишем значение | Panic |
Ограничения канала
Действие | Канал только на чтение | Канал только на запись |
|---|---|---|
Чтение | Ошибка компиляции | |
Запись | Ошибка компиляции | |
Закрытие | Ошибка компиляции |
Важные правила
Закрывает канал тот, кто в него пишет.
Если пишет несколько продюсеров, то закрывает тот, кто создал продюсеров.
Не закрытый канал держит ресурсы. Закрывать надо явно.
Паттерны
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)
}
}
