В двух предыдущих постах мы рассматривали Fanout и Fanin отдельно. Часто мы используем их вместе, когда у нас есть единый поток данных, в котором мы хотим работать с элементами индивидуально и можем делать это безопасно, используя параллелизм. Итак, мы разветвляемся на несколько рабочих потоков, а затем снова объединяемся в один поток.

Например, предположим, что у вас есть большой файл журнала. Вы можете разбить файл на части, позволяя каждому работнику одновременно работать с разными частями файла, а затем объединить результаты.

Если вы следили за двумя предыдущими постами, эта закономерность очевидна. Если вы не уверены, посмотрите ссылки выше.

// produce is simulating our single input as a channel
func produce() chan int {
    ch := make(chan int)
    go func() {
        for i := 0; i 



Существует функция Produce(), которая создает моделируемый входной поток чисел. 

Существует рабочая функция, которая работает с входным каналом до тех пор, пока не закончатся данные. Для каждого значения он «обрабатывает» входные данные (определяет, является ли значение нечетным или четным), а затем отправляет структуру результата в выходной канал. 

Обратите внимание, что когда каждый рабочий процесс завершает работу, он закрывает свой канал результатов. Это необходимо для предотвращения тупиковой ситуации, поскольку в противном случае операция fanin остановилась бы в ожидании дополнительных данных в канале.

Основной поток получает входной поток от продукта, затем запускает несколько рабочих процессов, предоставляя каждому рабочему канал свой собственный канал, куда он будет отправлять свои результаты.

Эти каналы результатов затем отправляются в операцию объединения. Для этого мы создаем канал для получения выходных данных, а затем запускаем горутину для каждого из рабочих каналов. Каждая горутина просто перебирает канал до тех пор, пока не закончатся данные, а затем завершается. Помните, что мы закрыли канал результатов в рабочем потоке, что позволяет циклу for завершиться

Обратите внимание, что мы используем WaitGroup для процесса раздачи. Это позволит нам узнать, когда все результаты всех каналов результатов были объединены в выходной канал. Когда это происходит, мы закрываем выходной канал, чтобы любой нижестоящий поток, потребляющий выходные данные, мог завершиться.

Со всеми данными в выходном канале основной поток может продолжить работу и отобразить результаты. Обратите внимание, что мы используем логический канал, чтобы предотвратить завершение основного потока, пока все не будет выполнено; в противном случае процесс будет завершен.

Обратите внимание, что существует другой способ объединения с помощью оператора select. Используемый здесь метод немного проще, поскольку мы можем увеличивать или уменьшать количество рабочих.

Также обратите внимание, что мы ничего не рассмотрели в отношении досрочного прекращения действия таких средств, как SIGTERM или SIGINT. Это добавляет немного больше сложности.

Как бы вы это реализовали? Существуют и другие реализации шаблона fanout/fanin. Пожалуйста, оставьте свои комментарии и мысли ниже?

Спасибо!

Код этого поста и всех постов этой серии можно найти здесь