Golang如何实现并发任务的结果聚合 演示Fan-in与Fan-out模式

package main

import (
    "fmt"
    "sync"
)

// worker 函数，计算数字的平方并将结果发送到输出 channel
func worker(id int, jobs <-chan int, results chan<- int) {
    for j := range jobs {
        fmt.Printf("worker %d processing job %d\n", id, j)
        results <- j * j
    }
}

// fanIn 函数，将多个输入 channel 的数据合并到一个输出 channel
func fanIn(inputChannels ...<-chan int) <-chan int {
    var wg sync.WaitGroup
    outputChannel := make(chan int)

    // 为每个输入 channel 启动一个 goroutine，将其数据复制到输出 channel
    for _, inputChannel := range inputChannels {
        wg.Add(1)
        go func(inputChannel <-chan int) {
            defer wg.Done()
            for n := range inputChannel {
                outputChannel <- n
            }
        }(inputChannel)
    }

    // 启动一个 goroutine，在所有输入 channel 都关闭后关闭输出 channel
    go func() {
        wg.Wait()
        close(outputChannel)
    }()

    return outputChannel
}

func main() {
    // 定义要处理的数字列表
    numbers := []int{1, 2, 3, 4, 5, 6, 7, 8, 9, 10}

    // 创建一个 jobs channel，用于发送要处理的数字
    jobs := make(chan int, len(numbers))
    // 创建一个 results channel，用于接收计算结果
    results := make(chan int, len(numbers))

    // 启动多个 worker goroutine (Fan-out)
    numWorkers := 3
    for w := 1; w <= numWorkers; w++ {
        go worker(w, jobs, results)
    }

    // 将数字发送到 jobs channel
    for _, number := range numbers {
        jobs <- number
    }
    close(jobs)

    // 从 results channel 收集结果 (Fan-in)
    //  这里先收集results channel到多个中间channel，然后再FanIn
    intermediateChannels := make([]<-chan int, numWorkers)
    for i := 0; i < numWorkers; i++ {
        intermediateChannels[i] = make(chan int, len(numbers)/numWorkers) // 假设每个worker大致处理均等数量的任务
        go func(i int, results <-chan int, out chan<- int) {
            defer close(out)
            count := 0
            for r := range results {
                if count%numWorkers == i { // 简单的轮询分配
                    out <- r
                }
                count++
            }
        }(i, results, intermediateChannels[i].(chan int)) // 类型断言，因为intermediateChannels是interface切片
    }
    // 调用 fanIn 函数将多个中间 channel 的数据合并到一个 channel
    mergedResults := fanIn(intermediateChannels...)

    // 打印计算结果
    for result := range mergedResults {
        fmt.Println(result)
    }
}