Go 之 管道

基本概念

声明和创建

// 声明一个管道
var ch chan int

// 创建无缓冲管道
ch1 := make(chan int)

// 创建有缓冲管道
ch2 := make(chan int, 10) // 缓冲区容量为10

管道操作

基本操作

ch := make(chan int)

// 发送数据到管道
ch <- 42

// 从管道接收数据
value := <-ch

// 检查管道是否关闭
value, ok := <-ch
if !ok {
    fmt.Println("管道已关闭")
}

无缓冲 vs 有缓冲管道

无缓冲管道（同步管道）

func unbufferedChannel() {
    ch := make(chan int) // 无缓冲
    
    go func() {
        fmt.Println("开始发送数据")
        ch <- 1 // 阻塞，直到有人接收
        fmt.Println("数据发送完成")
    }()
    
    time.Sleep(2 * time.Second)
    fmt.Println("开始接收数据")
    <-ch // 接收数据，发送方解除阻塞
    fmt.Println("数据接收完成")
}

有缓冲管道（异步管道）

func bufferedChannel() {
    ch := make(chan int, 2) // 缓冲区大小为2
    
    ch <- 1 // 不阻塞
    ch <- 2 // 不阻塞
    // ch <- 3 // 这会阻塞，因为缓冲区已满
    
    fmt.Println(<-ch) // 1
    fmt.Println(<-ch) // 2
}

实际使用示例

1. 基本的 goroutine 通信

func basicCommunication() {
    ch := make(chan string)
    
    go func() {
        time.Sleep(2 * time.Second)
        ch <- "Hello from goroutine!"
    }()
    
    msg := <-ch
    fmt.Println(msg) // Hello from goroutine!
}

2. 多个 goroutine 协作

func worker(id int, jobs <-chan int, results chan<- int) {
    for job := range jobs {
        fmt.Printf("Worker %d processing job %d\n", id, job)
        time.Sleep(time.Second)
        results <- job * 2
    }
}

func workerPool() {
    jobs := make(chan int, 100)
    results := make(chan int, 100)
    
    // 启动3个worker
    for w := 1; w <= 3; w++ {
        go worker(w, jobs, results)
    }
    
    // 发送任务
    for j := 1; j <= 5; j++ {
        jobs <- j
    }
    close(jobs)
    
    // 收集结果
    for a := 1; a <= 5; a++ {
        result := <-results
        fmt.Printf("Result: %d\n", result)
    }
}

3. 使用 select 处理多个管道

func selectExample() {
    ch1 := make(chan string)
    ch2 := make(chan string)
    
    go func() {
        time.Sleep(1 * time.Second)
        ch1 <- "from ch1"
    }()
    
    go func() {
        time.Sleep(2 * time.Second)
        ch2 <- "from ch2"
    }()
    
    for i := 0; i < 2; i++ {
        select {
        case msg1 := <-ch1:
            fmt.Println(msg1)
        case msg2 := <-ch2:
            fmt.Println(msg2)
        case <-time.After(3 * time.Second):
            fmt.Println("timeout")
            return
        }
    }
}

4. 管道方向（只读/只写）

// 只写管道
func producer(ch chan<- int) {
    for i := 0; i < 5; i++ {
        ch <- i
    }
    close(ch)
}

// 只读管道
func consumer(ch <-chan int) {
    for num := range ch {
        fmt.Println("Received:", num)
    }
}

func directionalChannels() {
    ch := make(chan int)
    go producer(ch)
    consumer(ch)
}

5. 使用 context 控制管道操作

func contextExample(ctx context.Context, ch chan int) {
    for {
        select {
        case <-ctx.Done():
            fmt.Println("Context cancelled")
            return
        case value := <-ch:
            fmt.Println("Received:", value)
        case <-time.After(1 * time.Second):
            fmt.Println("Timeout waiting for data")
        }
    }
}

高级模式

1. 扇出（Fan-out）

// 一个生产者，多个消费者
func fanOut(input <-chan int, outputs []chan int) {
    for value := range input {
        for _, output := range outputs {
            output <- value
        }
    }
    
    for _, output := range outputs {
        close(output)
    }
}

2. 扇入（Fan-in）

// 多个生产者，一个消费者
func fanIn(inputs ...<-chan int) <-chan int {
    output := make(chan int)
    
    var wg sync.WaitGroup
    for _, input := range inputs {
        wg.Add(1)
        go func(ch <-chan int) {
            defer wg.Done()
            for value := range ch {
                output <- value
            }
        }(input)
    }
    
    go func() {
        wg.Wait()
        close(output)
    }()
    
    return output
}

3. 管道管道（Pipeline）

func generator(nums ...int) <-chan int {
    out := make(chan int)
    go func() {
        for _, n := range nums {
            out <- n
        }
        close(out)
    }()
    return out
}

func square(in <-chan int) <-chan int {
    out := make(chan int)
    go func() {
        for n := range in {
            out <- n * n
        }
        close(out)
    }()
    return out
}

func pipelineExample() {
    // 设置管道：生成数字 -> 平方 -> 输出
    for n := range square(square(generator(1, 2, 3, 4))) {
        fmt.Println(n) // 输出 1, 16, 81, 256
    }
}

最佳实践和注意事项

1. 总是关闭管道：发送方负责关闭管道

2. 避免死锁：确保有对应的接收方

3. 使用 select 超时：避免永久阻塞

4. 管道所有权：明确哪个 goroutine 负责关闭管道

5. 使用缓冲管道：当生产者和消费者速度不匹配时

// 安全的管道使用模式
func safeChannelUsage() {
    ch := make(chan int, 10)
    done := make(chan struct{})
    
    // 生产者
    go func() {
        defer close(ch) // 生产者负责关闭
        for i := 0; i < 10; i++ {
            select {
            case ch <- i:
            case <-done:
                return
            }
        }
    }()
    
    // 消费者
    go func() {
        for value := range ch { // 自动检测关闭
            fmt.Println(value)
        }
        fmt.Println("Channel closed")
    }()
    
    time.Sleep(1 * time.Second)
    close(done) // 优雅停止
}