Go并发模式：详解Fan-Out（一生产者多消费者）

func fanOut(ch <-chan int, size, lag int) []chan int {
    // ... 实现细节
}

    cs := make([]chan int, size)
    for i, _ := range cs {
        // lag参数决定了通道的缓冲大小
        cs[i] = make(chan int, lag)
    }

    go func() {
        for i := range ch { // 从输入通道读取数据
            for _, c := range cs { // 将数据副本发送到所有输出通道
                c <- i
            }
        }
        // 当输入通道关闭且所有数据被读取完毕后，关闭所有输出通道
        for _, c := range cs {
            close(c)
        }
    }()

func fanOutUnbuffered(ch <-chan int, size int) []chan int {
    cs := make([]chan int, size)
    for i, _ := range cs {
        cs[i] = make(chan int) // 无缓冲通道
    }
    go func() {
        for i := range ch {
            for _, c := range cs {
                c <- i
            }
        }
        for _, c := range cs {
            close(c)
        }
    }()
    return cs
}

package main

import (
    "fmt"
    "time"
)

// producer 函数模拟一个数据生产者
// 它会生成指定数量的整数，并每秒发送一个
func producer(iters int) <-chan int {
    c := make(chan int)
    go func() {
        for i := 0; i < iters; i++ {
            c <- i
            time.Sleep(1 * time.Second) // 模拟生产耗时
        }
        close(c) // 生产完毕后关闭通道
    }()
    return c
}

// consumer 函数模拟一个数据消费者
// 它从输入通道读取数据并打印
func consumer(cin <-chan int) {
    for i := range cin {
        fmt.Printf("Consumer received: %d\n", i)
    }
    fmt.Println("Consumer finished.")
}

// fanOut 函数实现带缓冲的Fan-Out模式
// ch: 输入通道
// size: 输出通道的数量
// lag: 输出通道的缓冲大小
func fanOut(ch <-chan int, size, lag int) []chan int {
    cs := make([]chan int, size)
    for i := range cs {
        cs[i] = make(chan int, lag) // 创建带缓冲的输出通道
    }
    go func() {
        for i := range ch { // 从输入通道读取数据
            for _, c := range cs { // 将数据副本发送到所有输出通道
                c <- i
            }
        }
        // 输入通道关闭后，关闭所有输出通道
        for _, c := range cs {
            close(c)
        }
    }()
    return cs
}

// fanOutUnbuffered 函数实现无缓冲的Fan-Out模式
func fanOutUnbuffered(ch <-chan int, size int) []chan int {
    cs := make([]chan int, size)
    for i := range cs {
        cs[i] = make(chan int) // 创建无缓冲的输出通道
    }
    go func() {
        for i := range ch {
            for _, c := range cs {
                c <- i
            }
        }
        for _, c := range cs {
            close(c)
        }
    }()
    return cs
}

func main() {
    // 1. 创建一个生产者，生产10个数据
    c := producer(10)

    // 2. 使用fanOutUnbuffered函数创建3个输出通道
    // 尝试将 fanOutUnbuffered 替换为 fanOut(c, 3, 1) 或 fanOut(c, 3, 5)
    // 观察缓冲对行为的影响
    chans := fanOutUnbuffered(c, 3) 

    // 3. 启动3个消费者
    // 前两个消费者作为goroutine运行
    go consumer(chans[0])
    go consumer(chans[1])
    // 最后一个消费者在主goroutine中运行，阻塞主goroutine直到其完成
    consumer(chans[2]) 

    fmt.Println("Main goroutine finished.")
}