Знакомы ли вам паттерны fan-in/fan-out и pipeline? Что это такое?

Паттерны fan-in/fan-out и pipeline часто применяются для организации параллельной и конвейерной обработки данных в Go.

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

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

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

Пример простого pipeline с fan-out и fan-in:

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

func worker(in <-chan int) <-chan int {
    out := make(chan int)
    go func() {
        defer close(out)
        for n := range in {
            out <- n * n // например, возводим в квадрат
        }
    }()
    return out
}

func fanIn(channels ...<-chan int) <-chan int {
    var wg sync.WaitGroup
    out := make(chan int)
    output := func(c <-chan int) {
        defer wg.Done()
        for n := range c {
            out <- n
        }
    }
    wg.Add(len(channels))
    for _, c := range channels {
        go output(c)
    }
    go func() {
        wg.Wait()
        close(out)
    }()
    return out
}

func main() {
    nums := generator(1, 2, 3, 4, 5)

    // Fan-out: запускаем несколько воркеров
    w1 := worker(nums)
    w2 := worker(nums)

    // Fan-in: объединяем результаты
    for n := range fanIn(w1, w2) {
        fmt.Println(n)
    }
}

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