Fan-in: merge

Merge

func merge(chs ...<-chan *FileMeta) <-chan *FileMeta {
    out := make(chan *FileMeta)
    var wg sync.WaitGroup

    for _, ch := range chs {
        wg.Add(1)
        go func(ch <-chan *FileMeta) {
            defer wg.Done()
            for v := range ch {
                out <- v
            }
        }(ch)
    }

    go func() {
        wg.Wait()
        close(out)
    }()

    return out
}

Каждый входной канал получает свою горутину: она читает и пересылает в общий out. Третья горутина ждёт все передатчики через WaitGroup и закрывает out — иначе читатель merge никогда бы не узнал, что данные кончились.

Три роли

Обрати внимание на разделение ролей: передатчики читают входы, WaitGroup считает их, закрывающая горутина — единственный владелец close(out). Это тот самый вопрос «кто закрывает» из, решённый правильно: закрывает не один из отправителей, а отдельная горутина, знающая, когда все закончили.

Fan-out + fan-in = пайплайн

Fan-out и fan-in обычно живут в одном пайплайне: стадия конвейера загрузки читает вход, раздаёт работу воркерам (fan-out), собирает результаты в один канал (fan-in) и отдаёт следующей стадии. Следующая стадия — снова fan-out — и так цепочкой.

<!-- wp:course-hub/quiz {"title":"Проверьте себя","questions":[{"text":"Кто закрывает выходной канал out в merge?","options":[{"text":"Отдельная горутина после wg.Wait() — единственный владелец close(out)","correct":true,"comment":"Ни один передатчик не может закрывать: другой ещё отправляет — паника."},{"text":"Первый закончивший передатчик","correct":false},{"text":"Читатель","correct":false},{"text":"Производитель","correct":false}]},{"text":"Зачем обобщать merge на дженерики?","options":[{"text":"Один merge на все типы каналов — без копий mergeInt/mergeString","correct":true,"comment":"Параметр типа на каналах убирает дублирование."},{"text":"Это быстрее","correct":false},{"text":"Дженерики обязательны для каналов","correct":false},{"text":"Без них merge не работает","correct":false}]},{"text":"Читатель merge ушёл раньше, а передатчик шлёт в out. Что происходит?","options":[{"text":"Передатчик блокируется на out