Pipeline: возведение в квадрат

Решение

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

Подход

1. Читаем данные из input канала через range
2. Для каждого значения:
   • Вычисляем квадрат
   • Отправляем в output канал
3. Когда input закрывается, range завершается
4. Закрываем output канал (сигнал о завершении)

Реализация

func squarePipeline(input <-chan int, output chan<- int) {
    defer close(output)  // гарантирует закрытие output
    
    for num := range input {  // читаем из input пока он не закрыт
        output <- num * num   // отправляем квадрат в output
    }
}

Пошаговый пример

func main() {
    input := make(chan int)
    output := make(chan int)
    
    // Запускаем pipeline в background
    go squarePipeline(input, output)
    
    // Отправляем данные
    go func() {
        for i := 1; i <= 5; i++ {
            input <- i      // отправляем 1, 2, 3, 4, 5
        }
        close(input)        // сигнал: больше данных нет
    }()
    
    // Читаем результаты
    for result := range output {
        fmt.Println(result)  // выводит: 1, 4, 9, 16, 25
    }
}

Выполнение:

1. squarePipeline начинает ждать данных из input
2. Горутина отправляет 1 → pipeline вычисляет 1² = 1 → отправляет в output
3. Main читает 1
4. Горутина отправляет 2 → pipeline вычисляет 2² = 4 → отправляет в output
5. Main читает 4
... (повторяется для 3, 4, 5)
6. Горутина закрывает input
7. squarePipeline range завершается
8. defer close(output) закрывает output
9. Main range завершается (канал закрыт)

Полный код с примерами

package main

import "fmt"

func squarePipeline(input <-chan int, output chan<- int) {
    defer close(output)
    for num := range input {
        output <- num * num
    }
}

func main() {
    input := make(chan int)
    output := make(chan int)
    
    go squarePipeline(input, output)
    
    // Отправляем данные
    go func() {
        for i := 1; i <= 5; i++ {
            input <- i
        }
        close(input)
    }()
    
    // Читаем результаты
    for result := range output {
        fmt.Println(result)  // 1, 4, 9, 16, 25
    }
}

Вариант с буферизованными каналами

func main() {
    input := make(chan int, 5)    // буферизованный
    output := make(chan int, 5)   // буферизованный
    
    go squarePipeline(input, output)
    
    // Отправляем все сразу
    for i := 1; i <= 5; i++ {
        input <- i
    }
    close(input)
    
    // Читаем результаты
    for result := range output {
        fmt.Println(result)
    }
}

Цепочка pipelines (более сложный пример)

Можно создать несколько этапов обработки:

// Этап 1: возведение в квадрат
func square(input <-chan int, output chan<- int) {
    defer close(output)
    for num := range input {
        output <- num * num
    }
}

// Этап 2: умножение на 2
func double(input <-chan int, output chan<- int) {
    defer close(output)
    for num := range input {
        output <- num * 2
    }
}

func main() {
    // Создаём каналы
    input := make(chan int)
    middle := make(chan int)
    output := make(chan int)
    
    // Запускаем этапы
    go square(input, middle)    // input → square → middle
    go double(middle, output)   // middle → double → output
    
    // Отправляем данные
    go func() {
        for i := 1; i <= 5; i++ {
            input <- i
        }
        close(input)
    }()
    
    // Читаем результаты
    for result := range output {
        fmt.Println(result)  // (1²)*2=2, (2²)*2=8, (3²)*2=18, ...
    }
}

Pipeline с обработкой ошибок

type Result struct {
    Value int
    Error error
}

func squarePipelineWithError(input <-chan int, output chan<- Result) {
    defer close(output)
    
    for num := range input {
        if num < 0 {
            output <- Result{Error: fmt.Errorf("negative number: %d", num)}
        } else {
            output <- Result{Value: num * num}
        }
    }
}

func main() {
    input := make(chan int)
    output := make(chan Result)
    
    go squarePipelineWithError(input, output)
    
    go func() {
        for _, i := range []int{1, -2, 3, 4} {
            input <- i
        }
        close(input)
    }()
    
    for result := range output {
        if result.Error != nil {
            fmt.Printf("Ошибка: %v\n", result.Error)
        } else {
            fmt.Printf("Результат: %d\n", result.Value)
        }
    }
}

Важные детали

1. defer close(output) Гарантирует закрытие канала даже при паник:

func squarePipeline(input <-chan int, output chan<- int) {
    defer close(output)  // ВСЕГДА закроет output
    for num := range input {
        output <- num * num
    }
}

2. range автоматически заканчивается Когда канал закрывается:

for num := range input {
    // Когда input закрывается, range завершается
}

3. Нельзя отправлять в закрытый канал

ch := make(chan int)
close(ch)
ch <- 1  // ❌ PANIC: send on closed channel

Сравнение с другими подходами

❌ Без defer

func squarePipeline(input <-chan int, output chan<- int) {
    for num := range input {
        output <- num * num
    }
    close(output)  // может не выполниться при паник
}

✅ С defer

func squarePipeline(input <-chan int, output chan<- int) {
    defer close(output)  // ВСЕГДА выполнится
    for num := range input {
        output <- num * num
    }
}

Производительность

// Буферизованные каналы лучше для производительности
input := make(chan int, 100)    // буфер 100
output := make(chan int, 100)   // буфер 100

// Избегает блокировок, если отправитель опережает приёмник

Ключевые выводы

1. range на каналах — удобный способ итерировать до закрытия
2. defer close() — гарантирует закрытие канала
3. Pipelines — паттерн для обработки потоков данных
4. Можно цеплять — несколько этапов в одной цепочке
5. Не забывать закрывать — отправитель закрывает, приёмник завершает range

Eтот паттерн стандартный в Go для обработки конкурентных потоков данных.