Паттерны многопоточности: Конвейер, Fan-In, Fan-Out, Семафор

package main

import "fmt"

func main() {
    input := 1

    // пример 1
    // Вызываем функцию сложения, результат которой становится аргументом первого параметра функции умножения
    fmt.Println(multiply(add(input, 1), 2))

    // пример 2
    // можно поменять местами этапы, чтобы получить другой результат
    fmt.Println(add(multiply(input, 1), 2))
}

// add функция сложения
func add(x int, y int) int {
    return x + y
}

// multiply функция умножения
func multiply(x int, y int) int {
    return x * y
} 

package main

import "fmt"

// add принимает на вход сигнальный канал для прекращения работы и канал с входными данными для работы, 
// а возвращает канал, в который будет отправляться результат вычислений.
// На фоне будет запущена горутина, выполняющая вычисления до момента закрытия doneCh.
func add(doneCh chan struct{}, inputCh chan int) chan int {
    // канал с результатом
    addRes := make(chan int)

    // горутина, в которой добавляем к значению из inputCh единицу и отправляем результат в addRes
    go func() {
        // закрываем канал, когда горутина завершается
        defer close(addRes)

        // берём из канала inputCh значения, которые надо изменить
        for data := range inputCh {
            result := data + 1

            select {
            // если канал doneCh закрылся, выходим из горутины
            case <-doneCh:
                return
            // если doneCh не закрыт, отправляем результат вычисления в канал результата
            case addRes <- result:
            }
        }
    }()
    // возвращаем канал для результатов вычислений
    return addRes
}


// multiply принимает на вход сигнальный канал для прекращения работы и канал с входными данными для работы, 
// а возвращает канал, в который будет отправляться результат вычислений.
// На фоне будет запущена горутина, выполняющая вычисления до момента закрытия doneCh.
func multiply(doneCh chan struct{}, inputCh chan int) chan int {
    // канал с результатом
    multiplyRes := make(chan int)

    // горутина, в которой значение из inputCh умножаем на 2 и возвращаем в канал multiplyRes
    go func() {
        // закрываем канал multipleRes, когда горутина завершается
        defer close(multiplyRes)

        // берем из канала inputCh значения, которые надо изменить
        for data := range inputCh {
            // изменяем данные
            result := data * 2

            select {
            // если канал doneCh закрылся, выходим из горутины
            case <-doneCh:
                return
            // если doneCh не закрыт, отправляем результат вычисления в канал результата
            case multiplyRes <- result:
            }
        }
    }()

    // возвращаем канал для результатов вычислений
    return multiplyRes
}


// generator возвращает канал с данными
func generator(doneCh chan struct{}, input []int) chan int {
    // канал, в который будем отправлять данные из слайса
    inputCh := make(chan int)

    // горутина, в которой отправляем в канал  inputCh данные
    go func() {
        // как отправители закрываем канал, когда всё отправим
        defer close(inputCh)

        // перебираем все данные в слайсе
        for _, data := range input {
            select {
            // если doneCh закрыт, сразу выходим из горутины
            case <-doneCh:
                return
            // если doneCh не закрыт, кидаем в канал inputCh данные data
            case inputCh <- data:
            }
        }
    }()

    // возвращаем канал для данных
    return inputCh
}

func main() {
    // ваши данные в слайсе
    input := []int{1, 2, 3, 4, 5, 6, 7, 8}

    // канал для сигнала к выходу из горутины
    doneCh := make(chan struct{})
    // при завершении программы закрываем канал doneCh, чтобы все горутины тоже завершились
    defer close(doneCh)

    // получаем канал с данными с помощью генератора
    inputCh := generator(doneCh, input)

    // ваш конвейер, сначала работает add,  потом multiply
    resultCh := multiply(doneCh, add(doneCh, inputCh))

    // выводим результат
    for res := range resultCh {
        fmt.Println(res)
    }
} 

 addResults := add(doneCh, inputCh)
 resultCh := multiply(doneCh, addResults) 

package main

import (
    "fmt"
)

// generator функция из предыдущего примера, делает то же, что и делала
func generator(doneCh chan struct{}, input []int) chan int {
    inputCh := make(chan int)

    go func() {
        defer close(inputCh)

        for _, data := range input {
            select {
            case <-doneCh:
                return
            case inputCh <- data:
            }
        }
    }()

    return inputCh
}

// multiply функция из предыдущего примера, делает то же, что и делала
func multiply(doneCh chan struct{}, inputCh chan int) chan int {
    multiplyRes := make(chan int)

    go func() {
        defer close(multiplyRes)

        for data := range inputCh {
            result := data * 2

            select {
            case <-doneCh:
                return
            case multiplyRes <- result:
            }
        }
    }()
    return multiplyRes
}

// add функция из предыдущего примера, делает то же, что и делала
func add(doneCh chan struct{}, inputCh chan int) chan int {
    addRes := make(chan int)

    go func() {
        defer close(addRes)

        for data := range inputCh {
            // замедлим вычисление, как будто функция add требует больше вычислительных ресурсов
            time.Sleep(time.Second)

            result := data + 1

            

            select {
            case <-doneCh:
                return
            case addRes <- result:
            }
        }
    }()
    return addRes
}


// fanOut принимает канал данных, порождает 10 горутин
func fanOut(doneCh chan struct{}, inputCh chan int) []chan int {
    // количество горутин add
    numWorkers := 10
    // каналы, в которые отправляются результаты
    channels := make([]chan int, numWorkers)


    for i := 0; i < numWorkers; i++ {
        // получаем канал из горутины add
        addResultCh := add(doneCh, inputCh)
        // отправляем его в слайс каналов
        channels[i] = addResultCh
    }

    // возвращаем слайс каналов
    return channels
} 

// fanIn объединяет несколько каналов resultChs в один.
func fanIn(doneCh chan struct{}, resultChs ...chan int) chan int {
    // конечный выходной канал в который отправляем данные из всех каналов из слайса, назовём его результирующим
    finalCh := make(chan int)

    // понадобится для ожидания всех горутин
    var wg sync.WaitGroup

    // перебираем все входящие каналы
    for _, ch := range resultChs {
        // в горутину передавать переменную цикла нельзя, поэтому делаем так 
        chClosure := ch

        // инкрементируем счётчик горутин, которые нужно подождать
        wg.Add(1)

        go func() {
            // откладываем сообщение о том, что горутина завершилась
            defer wg.Done()

            // получаем данные из канала
            for data := range chClosure {
                select {
                // выходим из горутины, если канал закрылся
                case <-doneCh:
                    return
                // если не закрылся, отправляем данные в конечный выходной канал
                case finalCh <- data:
                }
            }
        }()
    }

    
    go func() {
        // ждём завершения всех горутин
        wg.Wait()
        // когда все горутины завершились, закрываем результирующий канал
        close(finalCh)
    }()

    // возвращаем результирующий канал
    return finalCh
} 

func main() {
    // слайс данных
    input := []int{1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15}

    // сигнальный канал для завершения горутин
    doneCh := make(chan struct{})
    // закрываем его при завершении программы
    defer close(doneCh)

    // канал с данными
    inputCh := generator(doneCh, input)

    // получаем слайс каналов из 10 рабочих add
    channels := fanOut(doneCh, inputCh)

    // а теперь объединяем десять каналов в один
    addResultCh := fanIn(doneCh, channels...)

    // передаём тот один канал в следующий этап обработки
    resultCh := multiply(doneCh, addResultCh)

    // выводим результаты расчетов из канала
    for res := range resultCh {
        fmt.Println(res)
    }
} 

package main

import (
    "fmt"
    "sync"
    "time"
)

// Semaphore структура семафора
type Semaphore struct {
    semaCh chan struct{}
}

// NewSemaphore создает семафор с буферизованным каналом емкостью maxReq
func NewSemaphore(maxReq int) *Semaphore {
    return &Semaphore {
        semaCh: make(chan struct{}, maxReq),
    }
}

// когда горутина запускается, отправляем пустую структуру в канал semaCh
func (s *Semaphore) Acquire() {
    s.semaCh <- struct{}{}
}

// когда горутина завершается, из канала semaCh убирается пустая структура
func (s *Semaphore) Release() {
    <-s.semaCh
} 

func main() {
    // чтобы дождаться всех горутин
    var wg sync.WaitGroup

    // создаём семафор емкостью 2: он будет пропускать только 2 горутины
    semaphore := NewSemaphore(2)

    // создаем 10 горутин
    for idx := 0; idx < 10; idx++ {
        wg.Add(1)

        // горутина в которую помещаем её порядковый номер
        go func(taskID int) {
            // отправляем в канал семафора пустую структуру
            semaphore.Acquire()

            // откладываем уменьшение счетчика в WaitGroup, когда завершится горутина
            defer wg.Done()

            // забираем из канала семафора пустую структуру, дав возможность запуститься другим горутинам
            defer semaphore.Release()

            log.Println("Запущен рабочий %d", taskID)

            time.Sleep(1 * time.Second)
        }(idx)
    }

    // ждём завершения всех горутин
    wg.Wait()
} 

package main

import (
    "fmt"
    "time"
)

// worker это наш рабочий, который принимает два канала:
// jobs - канал задач, это входные данные для обработки
// results - канал результатов, это результаты работы воркера
func worker(id int, jobs <-chan int, result chan<- int) {
    for j := range jobs {
        // для наглядности будем выводить какой рабочий начал работу и кго задачу
        fmt.Println("рабочий", id, "запущен задача", j)
        // немного замедлим выполнение рабочего
        time.Sleep(time.Second)
        // для наглядности выводим какой рабочий завершил какую задачу
        fmt.Println("рабочий", id, "закончил задача", j)
        // отправляем результат в канал результатов
        results <- j + 1
    }
}

func main() {
    // допустим у вас 5 задач, которые нужно выполнить
    const numJobs = 5
    // создаем буферизованный канал для принятия задач в воркер
    jobs := make(chan int, numJobs)
    // создаем буферизованный канал для отправки результатов
    results := make(chan int, numJobs)

    // создаем и запускаем 3 воркера, это и есть пул, 
    // передаем id, это для наглядности, канал задач и канал результатов
    for w := 1; w <= 3; w++ {
        go worker(w, jobs, results)
    }

    // в канал задач отправляем в канал задач какие-то данные
    // задач у нас 5, а воркера 3, значит одновременно решается только 3 задачи
    for j := 1; j <= numJobs; j++ {
        jobs <- j
    }
    // как вы помните, закрываем канал на стороне отправителя
    close(jobs)

    // забираем из канала результатов результаты ;)
    // можно присваивать переменной, или выводить на экран, но мы не будем 
    for a := 1; a <= numJobs; a++ {
        <-results
    }
}