> Как реализовать паттерн fan-out fan-in в Go с использованием горутин? (Go)

Короткий ответ

Fan-out fan-in - это паттерн параллельной обработки: несколько горутин (fan-out) читают из одного входного канала, а результаты объединяются в один выходной канал (fan-in). В Go реализуется через sync.WaitGroup для ожидания завершения воркеров и отдельную горутину-агрегатор, которая закрывает выходной канал после завершения всех воркеров. Ключевой момент - корректное закрытие каналов и обработка ошибок без deadlock.

Подробное объяснение

Паттерн состоит из трёх этапов:

1. Fan-out - запуск N горутин-воркеров, каждая из которых читает из общего входного канала jobs. Go гарантирует, что каждый элемент будет доставлен ровно одному воркеру.

2. Обработка - каждый воркер выполняет независимую операцию над полученным элементом и отправляет результат в общий выходной канал results.

3. Fan-in - агрегатор (обычно отдельная горутина) собирает результаты из выходного канала до его закрытия. Закрытие происходит после завершения всех воркеров.

Критический момент - закрытие выходного канала. Если закрыть его раньше, чем все воркеры закончат отправку, получим panic при отправке в закрытый канал. Если не закрыть вовсе - потребитель заблокируется навсегда. Поэтому используется sync.WaitGroup: каждый воркер вызывает wg.Done() после завершения, а отдельная горутина ждёт wg.Wait() и затем закрывает results.

Варианты реализации:

• Простой - все воркеры пишут в один канал, агрегатор читает из него.
• С буферизацией - буферизованные каналы снижают блокировки, но требуют аккуратного выбора размера буфера.
• С контекстом - context.Context для отмены при ошибках или таймаутах.
• С errgroup - golang.org/x/sync/errgroup упрощает обработку ошибок и отмену.

На практике

Для senior-позиции важно показать понимание trade-off:

• Количество воркеров - не равно числу элементов. Оптимально runtime.NumCPU() или настраиваемое значение.
• Ошибки - если один воркер упал, остальные должны получить сигнал остановки через context.
• Порядок результатов - fan-out fan-in не гарантирует порядок. Если порядок важен, нужна индексация результатов и сортировка после сбора.
• Буферизация - при большом потоке данных буфер канала results должен быть достаточным, иначе воркеры будут блокироваться на отправке.

Практический пример из реального бэкенда: обработка пачки HTTP-запросов к внешнему API, где каждый запрос независим, а результаты агрегируются в общий ответ.

Пример кода

GO
package main

import (
    "context"
    "fmt"
    "sync"
)

func fanOutFanIn(ctx context.Context, jobs <-chan int, workers int) <-chan int {
    results := make(chan int)
    var wg sync.WaitGroup

    // fan-out: запускаем воркеров
    for i := 0; i < workers; i++ {
        wg.Add(1)
        go func() {
            defer wg.Done()
            for job := range jobs {
                select {
                case <-ctx.Done():
                    return
                case results <- process(job):
                }
            }
        }()
    }

    // fan-in: закрываем results после завершения всех воркеров
    go func() {
        wg.Wait()
        close(results)
    }()

    return results
}

func process(n int) int {
    // имитация работы
    return n * 2
}

func main() {
    ctx := context.Background()
    jobs := make(chan int, 10)
    for i := 1; i <= 10; i++ {
        jobs <- i
    }
    close(jobs)

    results := fanOutFanIn(ctx, jobs, 3)
    for res := range results {
        fmt.Println(res)
    }
}

Как отвечать на собеседовании

1. Начни с определения паттерна и его назначения - параллельная обработка независимых задач.
2. Опиши структуру: входной канал, воркеры, выходной канал, агрегатор.
3. Обязательно упомяни sync.WaitGroup и почему без него будет deadlock или panic.
4. Покажи понимание edge cases: закрытие каналов, обработка ошибок, отмена через context.
5. Если спросят про производительность - расскажи про буферизацию, количество воркеров, runtime.GOMAXPROCS.
6. Предложи альтернативы: errgroup, worker pool библиотеки, channel с direction (send-only/receive-only).

Что проверяет интервьюер

• Понимание модели конкурентности Go: горутины, каналы, select.
• Умение работать с sync.WaitGroup и корректно закрывать каналы.
• Навыки проектирования конкурентных систем: баланс между параллелизмом и блокировками.
• Обработка ошибок и отмены операций.
• Способность объяснить trade-off между простотой и производительностью.

Типичные ошибки

• Закрытие канала в воркере - приводит к panic при отправке из других воркеров.
• Отсутствие wg.Wait() перед close(results) - deadlock у потребителя.
• Игнорирование context - при ошибке одного воркера остальные продолжают работать впустую.
• Слишком много воркеров - деградация производительности из-за переключения контекста.
• Забыли закрыть входной канал - воркеры будут вечно ждать новые задачи.
• Необработанные ошибки внутри воркера - результат теряется, а ошибка молча игнорируется.
• Использование общего sync.WaitGroup без Add перед запуском - гонка данных.