Go Channel并发通信的实现

ch := make(chan int)      // 无缓冲通道
ch := make(chan int, 5)   // 缓冲通道，容量为5

func producer(ch chan<- int) {  // 只能发送
    ch <- 42
}
func consumer(ch <-chan int) {  // 只能接收
    v := <-ch
}

select {
case msg := <-ch1:
    fmt.Println("收到 ch1:", msg)
case ch2 <- 42:
    fmt.Println("发送到 ch2")
case <-time.After(3 * time.Second):
    fmt.Println("超时了")
default:
    fmt.Println("所有通道都没准备好，立刻返回")
}

select {
case result := <-ch:
    return result
case <-time.After(5 * time.Second):
    return errors.New("超时")
}

for v := range ch {
    fmt.Println(v)
}
// 等价于：
for {
    v, ok := <-ch
    if !ok { break }  // 通道已关闭
    fmt.Println(v)
}

// ✅ 正确：发送方关闭
func sender(ch chan int) {
    for i := 0; i < 5; i++ {
        ch <- i
    }
    close(ch)  // 发送方关闭
}
// ❌ 错误：接收方关闭
func receiver(ch chan int) {
    v := <-ch
    close(ch)  // 危险！可能还有其他发送方
}
// ❌ 错误：重复关闭
close(ch)
close(ch)  // panic: close of closed channel

var sendOnly chan<- int = ch     // ✅ 编译通过
var recvOnly <-chan int = ch     // ✅ 编译通过
var both chan int = sendOnly      // ❌ 编译失败

// ❌ 泄漏：发送方退出后，接收方还在阻塞
func leak() {
    ch := make(chan int)
    go func() {
        for v := range ch {  // 永远等下去
            fmt.Println(v)
        }
    }()
    // 没有 close，也没有发送，goroutine 永远不退出
}

// ❌ 死锁：两个 goroutine 互相等对方
ch := make(chan int)
go func() { ch <- 1 }()     // 阻塞：没人接收
<-ch                         // 阻塞：没人发送
// fatal error: all goroutines are asleep - deadlock!

var ch chan int  // nil 通道，不是空通道！
ch <- 1          // 永久阻塞，不会 panic

ch := make(chan int, 3)
ch <- 1
ch <- 2
ch <- 3
ch <- 4  // 阻塞！容量已满

select {
case ch <- 1:  // 如果 ch 满了且没人收，这里永久阻塞
}
// 没有 default，没有超时，没有其他 case → 死锁

func workerPool(jobs <-chan int, results chan<- int, workers int) {
    var wg sync.WaitGroup
    for i := 0; i < workers; i++ {
        wg.Add(1)
        go func() {
            defer wg.Done()
            for job := range jobs {
                results <- process(job)
            }
        }()
    }
    wg.Wait()
    close(results)
}

// Fan-Out：一个入口，多个 worker 并行处理
func fanOut(input <-chan int) []<-chan int {
    outs := make([]<-chan int, 3)
    for i := range outs {
        ch := make(chan int)
        outs[i] = ch
        go func(out chan<- int) {
            for v := range input {
                out <- heavyWork(v)
            }
            close(out)
        }(ch)
    }
    return outs
}
// Fan-In：多个入口，汇聚到一个出口
func fanIn(ins ...<-chan int) <-chan int {
    out := make(chan int)
    var wg sync.WaitGroup
    for _, in := range ins {
        wg.Add(1)
        go func(ch <-chan int) {
            defer wg.Done()
            for v := range ch {
                out <- v
            }
        }(in)
    }
    go func() {
        wg.Wait()
        close(out)
    }()
    return out
}

func pipeline(input <-chan int) <-chan int {
    // Stage 1: 过滤
    filtered := filter(input, func(x int) bool { return x%2 == 0 })
    // Stage 2: 映射
    mapped := mapFunc(filtered, func(x int) int { return x * 2 })
    return mapped
}

func worker(ctx context.Context, jobs <-chan int) {
    for {
        select {
        case job, ok := <-jobs:
            if !ok { return }
            process(job)
        case <-ctx.Done():  // 收到取消信号，立即退出
            fmt.Println("worker 退出:", ctx.Err())
            return
        }
    }
}