Advanced Goroutines Patterns в Go: Fan-out, Fan-in и Pipelines

В языке Go одним из важнейших преимуществ является мощная поддержка многопоточности и параллелизма за счёт горутин и каналов. В этой статье подробно разберём три продвинутых шаблона работы с горутинами:

• Fan-out

• Fan-in

• Pipelines

Эти паттерны позволяют писать эффективный, масштабируемый и читабельный многопоточный код.

0. Как работают горутины под капотом в GO lang

В примере Fan-out из статьи, распределение работы происходит следующим образом:

Общий канал (jobs) используется как единая очередь задач, куда отправляются задания.

Запускается несколько воркеров (в примере — три воркера). Каждый воркер — это отдельная горутина, которая ожидает задачи из канала jobs.

Когда в канал поступает задача, один из свободных воркеров её забирает. Каналы работают по принципу FIFO (First In, First Out), и задачи передаются воркерам в порядке их поступления. Воркеры получают задачи именно тогда, когда они свободны.

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

Такой подход позволяет гибко распределять работу и использовать максимально эффективно доступные ресурсы (ядра CPU). Чем больше воркеров — тем больше параллелизм и выше потенциальная скорость обработки задач.

Распределение работы между воркерами происходит за счёт конструкции:

for job := range jobs {
    // выполнение задачи
}

Вот как это работает детально:

• Канал jobs общий для всех воркеров.

• Каждый воркер выполняет эту конструкцию (range jobs) в своей отдельной горутине.

• Когда в канал поступает задача, Go runtime автоматически передаёт её первому свободному воркеру, ожидающему данные из канала.

• Как только один из воркеров взял задачу из канала, другие воркеры эту задачу уже не увидят — она «выбрана» из канала и передана конкретному воркеру.

• Если все воркеры заняты, задача будет находиться в канале (ожидать), пока кто-то не освободится.

Таким образом, сам канал Go и конструкция range обеспечивают автоматическое и эффективное распределение задач между горутинами (воркерами).

Вот подробное объяснение, как это работает под капотом Go:

1. Каналы и их внутренняя реализация

В Go каналы (chan) — это механизм безопасного обмена данными между горутинами. Внутри они реализованы следующим образом:

Буферизация:
Канал может быть буферизованным (с фиксированным размером очереди) или небуферизованным (без очереди).

Небуферизованный канал: операция отправки (channel <- value) блокируется, пока кто-то не заберёт значение (<-channel).

Буферизованный канал: отправка блокируется только если канал полон, а получение — если пуст.

Очередь задач:
Канал представляет собой FIFO-очередь (First-In-First-Out), то есть данные, отправленные первыми, будут получены первыми.

2. Как происходит распределение задач горутинам?

Когда запускаются несколько горутин, каждая из них вызывает блокирующий оператор:

for job := range jobs {
    // выполнение задачи
}

• В этот момент каждая горутина пытается выполнить операцию чтения из канала.

• Если канал пуст, горутина «засыпает» (переходит в состояние ожидания).

• Когда в канал отправляется значение (например, jobs <- job), Go runtime пробуждает одну из ожидающих горутин, передавая ей значение.

• Если задач много и горутин несколько, задачи равномерно распределяются между свободными (ожидающими) горутинами в порядке отправки.

Таким образом, канал выступает как очередь задач, а горутины — как воркеры, забирающие задачи из этой очереди.

Роль Go runtime (scheduler)

Под капотом Go использует собственный планировщик (scheduler):

• Планировщик управляет горутинами и распределяет их по потокам ОС.

• Горутин обычно намного больше, чем потоков ОС, благодаря чему достигается высокая эффективность.

• Планировщик Go автоматически выбирает, какую горутину пробудить, когда данные становятся доступными для чтения из канала.

• При пробуждении горутины она ставится в очередь выполнения планировщика и получает процессорное время, как только оно освобождается.

Процесс выглядит так:

1. Горутина ждёт получения из канала.

2. Данные поступают в канал.

3. Go runtime выбирает и пробуждает ожидающую горутину.

4. Горутина берёт данные и продолжает работу.

Это обеспечивает эффективную и прозрачную работу многопоточности без сложной ручной синхронизации.

Итоговая схема взаимодействия:

Отправитель (main goroutine)
      │
      ▼
   Канал jobs ────────┐
    ▲    ▲    ▲       │
    │    │    │       │
goroutine goroutine goroutine 
(worker1) (worker2) (worker3)

Каждый раз, когда задача (job) отправляется в канал, она поступает первому свободному воркеру.

Таким образом, Go runtime и механизм каналов сами управляют распределением задач, что позволяет разработчику не думать о низкоуровневых деталях и полностью сосредоточиться на логике программы.

Как внутри под капотом работает планировшик на go

Планировщик горутин (goroutine scheduler) в Go — это ключевая часть среды выполнения Go, которая обеспечивает параллельность и эффективное выполнение горутин.

Как устроен планировщик горутин Go?

Планировщик Go называется GMP-моделью (Goroutine, Machine, Processor):
G (Goroutine) – это лёгкая пользовательская нить (user-space thread).
M (Machine) – это поток операционной системы (OS-thread).
P (Processor) – это логический процессор, представляющий контекст исполнения горутин (локальная очередь горутин, кеш).

Архитектура планировщика Go (GMP)

    G1   G2   G3 ... GN
     \    |    /
      \   |   /
         P1           P2 ... PN
          |             |
          M1            M2
          |             |
        Ядро ОС      Ядро ОС

Пояснение:

• Горутин может быть тысячи или даже миллионы.

• Горутин всегда больше, чем потоков ОС (M).

• Логические процессоры (P) распределяют горутины между доступными потоками ОС (M).

• Каждый логический процессор (P) имеет собственную локальную очередь горутин, которую он запускает по очереди.

• Потоки ОС (M) выполняют горутины, взятые из очередей логических процессоров (P).

Как горутина запускается и выполняется?

• Когда создаётся новая горутина (go func()), она помещается в очередь соответствующего процессора (P).

• Планировщик выбирает доступный поток ОС (M), который подключён к процессору (P), и начинает выполнять горутины из его локальной очереди.

• Если горутина блокируется (например, на канале, мьютексе или ожидании ввода-вывода), планировщик временно отцепляет поток ОС от процессора (P) и запускает другую горутину.

• Планировщик следит за тем, чтобы постоянно были заняты все доступные ядра процессора (CPU) и эффективно использовалось время выполнения.

Work Stealing (кража работы)

• Планировщик Go использует механизм «work stealing» (кража задач), чтобы эффективно балансировать нагрузку между процессорами (P):

• Если локальная очередь одного процессора (P) пуста, он может «украсть» горутины из очереди другого процессора (P), чтобы не простаивать.

• Это обеспечивает эффективное и равномерное распределение работы между всеми доступными ядрами процессора и потоками ОС.

Парковка и пробуждение горутин

Когда горутина блокируется (например, ожидает ввода-вывода или данных из канала):

• Go runtime помечает её как неактивную («паркует») и перестаёт выделять ей ресурсы CPU.

• Как только данные становятся доступны (например, в канале появляются данные), Go runtime пробуждает («разпарковывает») горутину, возвращает её в очередь на выполнение.

• Планировщик Go оперативно переключает потоки ОС и горутины, минимизируя время ожидания и простаивания CPU.

Взаимодействие с ОС и ядрами CPU

• Потоки ОС (M) напрямую взаимодействуют с ядрами CPU, именно они физически исполняют код.

• Go runtime автоматически увеличивает или уменьшает количество потоков ОС (M), чтобы оптимально использовать ресурсы CPU.

• Количество процессоров (P) по умолчанию равно количеству логических ядер CPU, но его можно регулировать через runtime.GOMAXPROCS.

Пример работы планировщика в действии:

Допустим, у нас 4 горутины и 2 логических процессора (P1 и P2):

G1 ──┐ 
G2 ──┼───▶ P1 ───▶ M1 ───▶ CPU1
     │
G3 ──┤ (work stealing)
G4 ──┴───▶ P2 ───▶ M2 ───▶ CPU2

Если P2 закончил работу быстрее, он попытается «украсть» работу (например, G2) у P1, тем самым балансируя нагрузку.

Итоги и преимущества GMP-модели

Планировщик горутин в Go:

• Автоматически балансирует нагрузку между ядрами CPU.

• Использует «work stealing» для эффективного распределения задач.

• Эффективно работает с блокирующими операциями (I/O, каналы, мьютексы).

• Позволяет запускать тысячи и миллионы горутин с минимальными затратами памяти и производительности.

Не требует ручного управления потоками и ядрами CPU.

Таким образом, Go runtime обеспечивает прозрачный, эффективный и высокопроизводительный механизм параллельного и конкурентного выполнения программ.

1. Что такое Fan-out и Fan-in?

Fan-out

Fan-out — это шаблон, при котором задачи из одного источника распределяются между несколькими воркерами. Основная цель — распараллелить работу и ускорить выполнение задачи.

Пример Fan-out:

func worker(id int, jobs <-chan int, results chan<- int) {
    for job := range jobs {
        fmt.Printf("Worker %d начал работу над задачей %d\n", id, job)
        time.Sleep(time.Second) // эмуляция работы
        results <- job * 2
        fmt.Printf("Worker %d завершил задачу %d\n", id, job)
    }
}

func main() {
    jobs := make(chan int, 10)
    results := make(chan int, 10)

    for w := 1; w <= 3; w++ {
        go worker(w, jobs, results)
    }

    for j := 1; j <= 9; j++ {
        jobs <- j
    }
    close(jobs)

    for a := 1; a <= 9; a++ {
        fmt.Println("Результат:", <-results)
    }
}

Fan-in

Fan-in — это обратный процесс, при котором результаты из нескольких каналов собираются в один.

Пример Fan-in:

func merge(cs ...<-chan int) <-chan int {
    out := make(chan int)
    var wg sync.WaitGroup

    output := func(c <-chan int) {
        for n := range c {
            out <- n
        }
        wg.Done()
    }

    wg.Add(len(cs))
    for _, c := range cs {
        go output(c)
    }

    go func() {
        wg.Wait()
        close(out)
    }()

    return out
}

func main() {
    c1 := make(chan int)
    c2 := make(chan int)

    go func() {
        for i := 1; i <= 5; i++ {
            c1 <- i
            time.Sleep(time.Millisecond * 200)
        }
        close(c1)
    }()

    go func() {
        for i := 6; i <= 10; i++ {
            c2 <- i
            time.Sleep(time.Millisecond * 300)
        }
        close(c2)
    }()

    for n := range merge(c1, c2) {
        fmt.Println("Получено:", n)
    }
}

2. Pipelines (Конвейеры)

Pipeline — это последовательность этапов обработки данных, где выход одного этапа становится входом для другого.

Пример Pipeline:

func gen(nums ...int) <-chan int {
    out := make(chan int)
    go func() {
        for _, n := range nums {
            out <- n
        }
        close(out)
    }()
    return out
}

func square(in <-chan int) <-chan int {
    out := make(chan int)
    go func() {
        for n := range in {
            out <- n * n
        }
        close(out)
    }()
    return out
}

func main() {
    for n := range square(square(gen(2, 3, 4))) {
        fmt.Println(n)
    }
}

3. Реальные примеры применения

Fan-out

• Обработка запросов веб-сервера: распределение HTTP-запросов между несколькими горутинами для ускорения ответа.

• Загрузка данных: одновременная загрузка файлов или ресурсов с нескольких URL.

Fan-in

• Агрегация логов: сбор данных из нескольких сервисов в единый канал для обработки и сохранения.

• Сбор статистики: получение метрик из нескольких источников и объединение их в единую систему мониторинга.

Pipelines

• Обработка изображений: загрузка, обработка (например, изменение размера или фильтрация), сохранение.

• ETL-процессы: извлечение данных из базы, трансформация, и последующая загрузка данных в другую базу или систему аналитики.

Заключение

Используя паттерны fan-out, fan-in и pipelines в Go, вы можете писать эффективные и производительные многопоточные приложения. Эти подходы обеспечивают чистоту кода, лёгкость масштабирования и гибкость для решения сложных задач.