如何在Golang中实现fan-in fan-out模式_Golang fan-in fan-out模式实践方法汇总

func fetch(url string, ch chan<- string) {
    resp, _ := http.Get(url)
    ch <- fmt.Sprintf("fetched %s: %d", url, resp.StatusCode)
}
<p>func main() {
urls := []string{"<a href="https://www.php.cn/link/374cad868cb62202053d308252bc4040">https://www.php.cn/link/374cad868cb62202053d308252bc4040</a>", "<a href="https://www.php.cn/link/5f46e3006c4072122784b2adcf7bb10e">https://www.php.cn/link/5f46e3006c4072122784b2adcf7bb10e</a>", "<a href="https://www.php.cn/link/44856cd0e9468bc2674a05c05210a144">https://www.php.cn/link/44856cd0e9468bc2674a05c05210a144</a>"}
resultCh := make(chan string, len(urls))</p><pre class='brush:php;toolbar:false;'>// Fan-out: 每个URL启动一个goroutine
for _, url := range urls {
    go fetch(url, resultCh)
}

// 收集所有结果
for i := 0; i < len(urls); i++ {
    fmt.Println(<-resultCh)
}

func worker(tasks <-chan int, results chan<- int, id int) {
    for num := range tasks {
        time.Sleep(time.Millisecond * 100) // 模拟耗时操作
        results <- num * num
        fmt.Printf("worker %d processed %d\n", id, num)
    }
}
<p>func main() {
tasks := make(chan int, 10)
results := make(chan int, 10)</p><pre class='brush:php;toolbar:false;'>// 启动3个worker
for i := 0; i < 3; i++ {
    go worker(tasks, results, i)
}

// 发送任务
for i := 1; i <= 5; i++ {
    tasks <- i
}
close(tasks)

// 收集结果
for i := 0; i < 5; i++ {
    fmt.Println("result:", <-results)
}

func fanIn(channels ...<-chan string) <-chan string {
    out := make(chan string)
    for _, ch := range channels {
        go func(c <-chan string) {
            for val := range c {
                out <- val
            }
        }(ch)
    }
<pre class='brush:php;toolbar:false;'>// 所有goroutine启动后关闭out（注意：此处简化处理）
go func() {
    for _, ch := range channels {
        for range ch {}
    }
    close(out)
}()

return out

func generate(nums ...int) <-chan int {
    out := make(chan int)
    go func() {
        for _, n := range nums {
            out <- n
        }
        close(out)
    }()
    return out
}
<p>func square(in <-chan int) <-chan int {
out := make(chan int)
go func() {
for n := range in {
out <- n * n
}
close(out)
}()
return out
}</p><p>func main() {
// Fan-out: 分发任务
nums := generate(1, 2, 3, 4, 5)</p><pre class='brush:php;toolbar:false;'>// 多个worker并行处理
c1 := square(nums)
c2 := square(nums)

// Fan-in: 合并结果
merged := merge(c1, c2)

// 输出结果
for v := range merged {
    fmt.Println(v)
}

output := func(c <-chan int) {
    for n := range c {
        out <- n
    }
    wg.Done()
}

wg.Add(len(cs))
for _, c := range cs {
    go output(c)
}

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