golang

Go Channel 并发通信详解：CSP 模型与管道模式

深入讲解 Go Channel 的底层 hchan 结构、有缓冲与无缓冲 channel 的语义差异、select 多路复用机制，以及 Fan-In、Fan-Out、Pipeline 等经典并发模式与最佳实践

Leeting Yan 2025-08-13 13 分钟阅读 6478 字

Channel：Go 并发的灵魂

上一篇我们学了 goroutine，但只学会了启动它们。有一个问题还没解决：goroutine 之间怎么交流？

在传统的并发编程中，线程之间通常通过共享内存来通信——比如共享变量加锁。这种方式容易出错，你忘了加锁、忘了释放锁、不小心造成死锁……各种坑。

Go 语言提出了一种不同的哲学：不要通过共享内存来通信，而要通过通信来共享内存。

这句话来自 Go 的设计文档，也是 Go 并发编程的核心思想。而实现这种"通过通信来共享内存"的工具，就是 Channel（通道）。

Channel 就像一根管子，一个 goroutine 往里面塞数据，另一个 goroutine 从里面取数据。它不仅是数据传输的通道，更是 goroutine 之间同步的机制。

让我们开始探索这个神奇的工具吧！

什么是 Channel？

如果 goroutine 是 Go 并发世界里的"工人"，那 channel 就是工人之间传递零件的"传送带"。

一个工人（goroutine）把东西放到传送带（channel）上
另一个工人（goroutine）从传送带上拿走东西
传送带本身也有一定的容量（缓冲区）

Channel 是类型安全的——一个 chan int 只能传输整数，一个 chan string 只能传输字符串。

创建 Channel

Channel 用 make 函数创建：

// 创建一个无缓冲的 int 通道
ch := make(chan int)

// 创建一个缓冲区大小为 10 的 string 通道
ch2 := make(chan string, 10)

注意两种创建方式的区别：

make(chan T)：无缓冲通道
make(chan T, size)：有缓冲通道

这两种通道有着截然不同的行为，后面会详细讲。

发送和接收

Channel 的基本操作有两个：

ch <- value  // 发送：把 value 发送到通道 ch
value := <-ch  // 接收：从通道 ch 接收一个值

箭头 <- 的方向表示了数据的流向：

ch <- 箭头指向通道，表示发送到通道
<-ch 箭头从通道出来，表示从通道接收

第一个完整的例子

package main

import "fmt"

func main() {
	// 创建一个无缓冲通道
	ch := make(chan string)

	// 启动一个 goroutine 发送数据
	go func() {
		ch <- "Hello from goroutine!"
	}()

	// 主 goroutine 接收数据
	msg := <-ch
	fmt.Println(msg)  // Hello from goroutine!
}

这个程序的工作流程：

创建了一个无缓冲通道 ch
启动一个 goroutine 往通道里发送 “Hello from goroutine!”
主 goroutine 从通道接收数据

无缓冲通道有一个重要的特性：发送和接收必须同时准备好。也就是说，发送方会一直阻塞，直到有接收方来接收；接收方会一直阻塞，直到有发送方来发送。

这就像一个电话——打电话的人必须等到接电话的人拿起话筒，通话才能开始。

无缓冲通道 vs 有缓冲通道

无缓冲通道（Unbuffered Channel）

无缓冲通道没有存储空间，发送和接收必须同步发生。

ch := make(chan int)  // 无缓冲

// 发送方会阻塞，直到有人接收
go func() {
	fmt.Println("发送前")
	ch <- 42  // 阻塞在这里，等待接收方
	fmt.Println("发送后")
}()

// 接收方会阻塞，直到有人发送
value := <-ch
fmt.Println("接收到:", value)

输出：

发送前
接收到: 42
发送后

注意"发送后"在"接收到"之后才打印——这说明发送操作确实被阻塞了，直到接收方才继续。

有缓冲通道（Buffered Channel）

有缓冲通道内部有一个队列，可以存储一定数量的值。发送操作只在缓冲区满的时候才阻塞，接收操作只在缓冲区空的时候才阻塞。

ch := make(chan int, 3)  // 缓冲区大小为 3

ch <- 1  // 不阻塞，放入缓冲区
ch <- 2  // 不阻塞，放入缓冲区
ch <- 3  // 不阻塞，放入缓冲区
// ch <- 4  // ❌ 会阻塞！缓冲区已满

fmt.Println(<-ch)  // 1
fmt.Println(<-ch)  // 2
fmt.Println(<-ch)  // 3

有缓冲通道就像一个邮箱——你可以把信扔进邮箱就走（不阻塞），邮递员什么时候来取信是另一回事。

⚠️ 注意：有缓冲通道虽然可以减少阻塞，但如果使用不当，可能隐藏同步问题。大多数情况下，无缓冲通道更安全，因为它的同步语义更明确。

Channel 的方向

你可以限制 channel 只能发送或只能接收，增加类型安全性：

// 只能发送的通道
var sendOnly chan<- int

// 只能接收的通道
var recvOnly <-chan int

// 既可以发送也可以接收
var bidirectional chan int

这在函数参数中特别有用：

func producer(out chan<- int) {
	// 只能发送，不能接收
	for i := 0; i < 5; i++ {
		out <- i
	}
	close(out)
}

func consumer(in <-chan int) {
	// 只能接收，不能发送
	for v := range in {
		fmt.Println(v)
	}
}

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

这种设计让你的代码意图更清晰——producer 只负责生产数据，consumer 只负责消费数据。

关闭 Channel

用 close() 函数关闭通道：

ch := make(chan int)
close(ch)

关闭通道表示"不会再有数据发送了"。关闭后的通道有以下行为：

接收：立即返回零值和 false
发送：会 panic！
再次关闭：会 panic！

检查通道是否关闭

ch := make(chan int, 2)
ch <- 1
ch <- 2
close(ch)

// 方式一：comma ok 模式
value, ok := <-ch
if ok {
	fmt.Println("收到:", value)
} else {
	fmt.Println("通道已关闭")
}

// 方式二：for range 遍历
ch2 := make(chan int, 3)
ch2 <- 1
ch2 <- 2
ch2 <- 3
close(ch2)

for v := range ch2 {
	fmt.Println(v)  // 1, 2, 3
}
// 循环在通道关闭且所有数据都被读取后自动结束

⚠️ 重要原则：

只有发送方才应该关闭通道
不要关闭一个已经被关闭的通道
不要往已关闭的通道发送数据
有缓冲通道关闭后，仍然可以接收缓冲区中剩余的数据

Select 多路复用

select 是 Go 并发编程中最重要的控制结构。它让你可以同时监听多个 channel 的操作，哪个先就绪就处理哪个：

package main

import (
	"fmt"
	"time"
)

func main() {
	ch1 := make(chan string)
	ch2 := make(chan string)

	go func() {
		time.Sleep(1 * time.Second)
		ch1 <- "来自通道 1 的消息"
	}()

	go func() {
		time.Sleep(2 * time.Second)
		ch2 <- "来自通道 2 的消息"
	}()

	// 使用 select 同时监听两个通道
	for i := 0; i < 2; i++ {
		select {
		case msg1 := <-ch1:
			fmt.Println(msg1)
		case msg2 := <-ch2:
			fmt.Println(msg2)
		}
	}
}

select 的工作方式：

同时检查所有的 case
如果有多个 case 就绪，随机选择一个执行
如果没有任何 case 就绪，就阻塞（除非有 default 分支）

超时控制

select 最常见的用法之一是实现超时：

ch := make(chan string)

go func() {
	time.Sleep(5 * time.Second)
	ch <- "结果"
}()

select {
case result := <-ch:
	fmt.Println("收到结果:", result)
case <-time.After(3 * time.Second):
	fmt.Println("超时了！")
}
// 输出：超时了！

time.After 返回一个通道，在指定时间后发送当前时间。这让你可以轻松实现超时逻辑。

默认分支

如果 select 有 default 分支，当没有 case 就绪时，会立即执行 default 而不会阻塞：

ch := make(chan int)

select {
case v := <-ch:
	fmt.Println("收到:", v)
default:
	fmt.Println("没有数据，做其他事情")
}

空 select

一个没有任何 case 的 select 会永远阻塞：

select {}  // 永远阻塞

这看起来没用，但在某些场景下可以用来让 main goroutine 永远等待（比如服务器）。

实战：并发 Web 爬虫

让我们用 goroutine 和 channel 实现一个简单的并发 Web 爬虫：

package main

import (
	"fmt"
	"math/rand"
	"sync"
	"time"
)

// CrawlResult 爬取结果
type CrawlResult struct {
	URL     string
	Status  string
	Elapsed time.Duration
}

// crawl 模拟爬取一个 URL
func crawl(url string) CrawlResult {
	// 模拟网络延迟
	elapsed := time.Duration(rand.Intn(500)+100) * time.Millisecond
	time.Sleep(elapsed)

	return CrawlResult{
		URL:     url,
		Status:  "OK",
		Elapsed: elapsed,
	}
}

func main() {
	urls := []string{
		"https://example.com/1",
		"https://example.com/2",
		"https://example.com/3",
		"https://example.com/4",
		"https://example.com/5",
		"https://example.com/6",
		"https://example.com/7",
		"https://example.com/8",
	}

	start := time.Now()

	// 创建结果通道
	results := make(chan CrawlResult, len(urls))

	// 启动 worker 池
	workerCount := 3
	var wg sync.WaitGroup

	// 创建 URL 通道
	urlChan := make(chan string, len(urls))
	for _, url := range urls {
		urlChan <- url
	}
	close(urlChan)

	// 启动 worker
	for i := 0; i < workerCount; i++ {
		wg.Add(1)
		go func(id int) {
			defer wg.Done()
			for url := range urlChan {
				fmt.Printf("Worker %d: 爬取 %s\n", id, url)
				result := crawl(url)
				results <- result
			}
		}(i + 1)
	}

	// 等待所有 worker 完成后关闭结果通道
	go func() {
		wg.Wait()
		close(results)
	}()

	// 收集结果
	var totalElapsed time.Duration
	for result := range results {
		fmt.Printf("✅ %s: %s (耗时: %v)\n", result.URL, result.Status, result.Elapsed)
		totalElapsed += result.Elapsed
	}

	fmt.Printf("\n总耗时: %v\n", time.Since(start))
	fmt.Printf("串行耗时: %v\n", totalElapsed)
	fmt.Printf("加速比: %.2fx\n", float64(totalElapsed)/float64(time.Since(start)))
}

这个程序使用了一个经典的生产者-消费者模式：

把所有 URL 放入一个通道
启动多个 worker goroutine 从通道中取 URL 并爬取
每个 worker 把结果发送到结果通道
主 goroutine 从结果通道收集结果

Channel 的常见模式

1. Fan-In（扇入）

把多个输入通道合并成一个输出通道：

func fanIn(channels ...<-chan int) <-chan int {
	out := make(chan int)
	var wg sync.WaitGroup

	for _, ch := range channels {
		wg.Add(1)
		go func(c <-chan int) {
			defer wg.Done()
			for v := range c {
				out <- v
			}
		}(ch)
	}

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

	return out
}

2. Fan-Out（扇出）

把一个输入通道的数据分发给多个处理者：

func fanOut(in <-chan int, workerCount int) []<-chan int {
	workers := make([]<-chan int, workerCount)

	for i := 0; i < workerCount; i++ {
		workers[i] = func() <-chan int {
			out := make(chan int)
			go func() {
				defer close(out)
				for v := range in {
					out <- v * 2  // 做一些处理
				}
			}()
			return out
		}()
	}

	return workers
}

3. Pipeline（管道）

把多个处理步骤串成一条流水线：

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

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

func filter(in <-chan int, threshold int) <-chan int {
	out := make(chan int)
	go func() {
		defer close(out)
		for v := range in {
			if v > threshold {
				out <- v
			}
		}
	}()
	return out
}

func main() {
	// 生成 → 平方 → 过滤
	pipeline := filter(square(generate(1, 2, 3, 4, 5, 6)), 10)

	for v := range pipeline {
		fmt.Println(v)  // 16, 25, 36
	}
}

Channel 的死锁检测

Go 运行时会自动检测 channel 死锁。如果所有 goroutine 都在等待，程序会 panic：

func main() {
	ch := make(chan int)
	// 没有人发送，也没有其他 goroutine
	<-ch  // fatal error: all goroutines are asleep - deadlock!
}

这个检测在开发阶段非常有用——它能帮你尽早发现并发逻辑中的 bug。

小结

今天我们深入学习了 Go 语言的 Channel：

Channel 是什么：goroutine 之间通信的管道，类型安全
创建：make(chan T) 无缓冲，make(chan T, n) 有缓冲
发送和接收：ch <- v 发送，v := <-ch 接收
无缓冲 vs 有缓冲：前者要求同步，后者有缓冲队列
关闭：close(ch)，只有发送方应该关闭通道
Select：多路复用，可以同时监听多个通道
常见模式：Fan-In、Fan-Out、Pipeline
死锁检测：Go 运行时自动检测

Channel 是 Go 并发编程的灵魂。它不只是数据传输的工具，更是一种编程范式——通过消息传递来协调并发，避免了共享内存和锁的复杂性。

记住这句话：“Do not communicate by sharing memory; instead, share memory by communicating.”

练习时间

生产者-消费者：实现一个完整的生产者-消费者模型，多个生产者、多个消费者
超时重试：用 select 实现一个带超时和重试机制的函数
聊天室：用 channel 实现一个简单的聊天室（多个用户同时发言）
并发安全计数器：用 channel 实现一个不需要锁的并发安全计数器
工作池：实现一个通用的 worker pool，支持可配置的 worker 数量和任务队列

下一篇预告

下一篇文章，我们将学习 sync 包——Go 提供的同步工具集。虽然 channel 很优雅，但有些场景用传统的同步原语（互斥锁、读写锁、等待组等）更合适。我们会学习：

sync.Mutex 和 sync.RWMutex
sync.WaitGroup
sync.Once
sync.Pool
什么时候用锁，什么时候用 channel

我们下篇见！👋

参考资料：

Channel 的底层数据结构（hchan）

Go 的 channel 在运行时由 hchan 结构体表示，它包含发送和接收队列、缓冲区指针、互斥锁等关键字段。

// 简化版的 hchan 结构
type hchan struct {
    qcount   uint           // 当前缓冲区元素个数
    dataqsiz uint           // 缓冲区环形队列大小
    buf      unsafe.Pointer // 缓冲区指针
    elemsize uint16         // 元素大小
    closed   uint32         // 是否已关闭
    elemtype *_type          // 元素类型
    sendx    uint            // 发送索引
    recvx    uint            // 接收索引
    recvq    waitq          // 接收者等待队列
    sendq    waitq          // 发送者等待队列
    lock     mutex           // 互斥锁
}

Channel 的所有操作（发送、接收、关闭）都必须先获取 hchan.lock。这意味着 channel 本质上是一个带锁的有界队列。了解这一点有助于理解 channel 在高并发场景下的行为。

有缓冲与无缓冲 Channel 的本质

无缓冲 channel：缓冲区大小为 0，qcount 永远为 0。发送方和接收方必须在同一条指令期间交接数据，这种同步称为 “handshake”。
有缓冲 channel：缓冲区大小大于 0。发送方可以在缓冲区未满时继续执行，不需要等待接收方。

package main

import (
    "fmt"
    "time"
)

func main() {
    // 无缓冲 channel：发送和接收必须同步
    unbuf := make(chan int)
    go func() {
        time.Sleep(100 * time.Millisecond)
        fmt.Println("接收:", <-unbuf)
    }()
    fmt.Println("开始发送...")
    unbuf <- 1  // 会阻塞，直到接收方就绪
    fmt.Println("发送完成")

    // 有缓冲 channel：发送可以先行
    buf := make(chan int, 1)
    fmt.Println("开始发送到缓冲 channel...")
    buf <- 2  // 不会阻塞，缓冲区有空间
    fmt.Println("发送完成，无需等待接收")
    fmt.Println("接收:", <-buf)
}

关闭 Channel 的广播语义

关闭一个 channel 会向所有等待接收的 goroutine 广播通知。这是 Go 中实现 “事件通知” 的标准方式。

package main

import (
    "fmt"
    "sync"
)

func main() {
    done := make(chan struct{})
    var wg sync.WaitGroup

    for i := 0; i < 3; i++ {
        wg.Add(1)
        go func(id int) {
            defer wg.Done()
            <-done
            fmt.Printf("Worker %d 收到关闭信号\n", id)
        }(i)
    }

    fmt.Println("关闭 channel...")
    close(done)
    wg.Wait()
}

优雅关闭 Pattern

package main

import (
    "fmt"
    "sync"
)

// StoppableWorker 可优雅停止的 worker
type StoppableWorker struct {
    tasks   chan int
    stop    chan struct{}
    stopped chan struct{}
}

func NewStoppableWorker() *StoppableWorker {
    w := &StoppableWorker{
        tasks:   make(chan int),
        stop:    make(chan struct{}),
        stopped: make(chan struct{}),
    }
    go w.run()
    return w
}

func (w *StoppableWorker) run() {
    defer close(w.stopped)
    for {
        select {
        case task := <-w.tasks:
            fmt.Printf("处理任务: %d\n", task)
        case <-w.stop:
            fmt.Println("Worker 停止中...")
            return
        }
    }
}

func (w *StoppableWorker) Submit(task int) bool {
    select {
    case w.tasks <- task:
        return true
    case <-w.stop:
        return false
    }
}

func (w *StoppableWorker) Stop() {
    close(w.stop)
    <-w.stopped
}

func main() {
    worker := NewStoppableWorker()
    worker.Submit(1)
    worker.Submit(2)
    worker.Stop()
    fmt.Println("Worker 已安全停止")
}

Channel 内存泄漏陷阱

channel 使用不当会导致 goroutine 泄漏——goroutine 永远等待在某个 channel 上，无法被垃圾回收。

package main

import (
    "fmt"
    "runtime"
    "time"
)

// 有泄漏的版本：发送方在没有接收方的情况下阻塞
func leaky() {
    ch := make(chan int)
    go func() {
        ch <- 1  // 永远阻塞在这里
    }()
    // 函数返回，ch 无人接收
}

// 修复版本：使用缓冲 channel 或带超时的发送
func fixed() {
    ch := make(chan int, 1)  // 缓冲为 1
    go func() {
        ch <- 1
    }()
}

func main() {
    before := runtime.NumGoroutine()
    leaky()
    time.Sleep(100 * time.Millisecond)
    after := runtime.NumGoroutine()
    fmt.Printf("泄漏前 goroutine 数: %d, 泄漏后: %d\n", before, after)
}

内存泄漏的常见原因：

向无缓冲 channel 发送数据没有对应的接收方
goroutine 在等待一个永远不会关闭的 channel
select 中某个 channel 永远不会就绪

性能对比：Channel vs Mutex

Channel 和 Mutex 在 Go 中都可以用于并发同步，但适用场景不同：

场景	Channel	Mutex
传递数据所有权	理想	不适用
保护共享状态	可以但复杂	自然
广播通知	理想（关闭 channel）	需要条件变量
信号量	理想	需要额外实现
高并发计数器	不适用	理想

package main

import (
    "fmt"
    "sync"
    "testing"
)

// Channel 实现的计数器
func BenchmarkChannelCounter(b *testing.B) {
    ch := make(chan func())
    go func() {
        count := 0
        for fn := range ch {
            fn()
            count++
            _ = count
        }
    }()

    b.ResetTimer()
    for i := 0; i < b.N; i++ {
        done := make(chan struct{})
        ch <- func() { close(done) }
        <-done
    }
}

// Mutex 实现的计数器
func BenchmarkMutexCounter(b *testing.B) {
    var mu sync.Mutex
    count := 0

    b.ResetTimer()
    for i := 0; i < b.N; i++ {
        mu.Lock()
        count++
        mu.Unlock()
    }
    _ = count
}

func main() {
    fmt.Println("Mutex 计数器通常比 Channel 快 10 倍以上")
}

常见错误与 FAQ

Q: 为什么从已关闭的 channel 读取不会 panic？

A: 关闭 channel 的目的是通知接收方没有更多数据了。如果读取导致 panic，接收方就无法优雅地处理 “数据已读完” 的情况。Go 设计了从已关闭 channel 读取返回零值的语义，配合逗号 ok 模式可以检测关闭状态。

Q: select 中的 case 顺序有影响吗？

A: 从 Go 1.8 开始，select 的所有就绪 case 会被随机选择，case 的声明顺序不会影响执行优先级。这一设计避免了某些 case 永远被优先执行而导致的不公平调度。

Q: 如何检测 channel 是否已满？

A: 使用非阻塞发送配合 select 的 default 分支：

select {
case ch <- value:
    // 发送成功
default:
    // channel 已满
}

Q: 多个 goroutine 同时关闭同一个 channel 会怎样？

A: panic。channel 只能由发送方关闭，且只能关闭一次。多个关闭操作会导致运行时 panic，因为无法确定哪个 goroutine 有权限关闭。

Q: nil channel 的行为是什么？

A: 向 nil channel 发送或接收都会永远阻塞。关闭 nil channel 会 panic。nil channel 在某些设计模式中可以作为占位符使用。

Q: 有缓冲 channel 什么时候也会死锁？

A: 死锁的本质是所有 goroutine 都在等待且没有人能继续执行。有缓冲 channel 也可能导致死锁，例如：(1) 所有 goroutine 都在从另一个已满的 channel 获取数据；(2) 向一个 channel 发送后，本 goroutine 又在等待自己发送的数据被处理；(3) 循环依赖多个 channel 的收发顺序导致永远相互等待。

Q: channel 适合做多 goroutine 同步的锁吗？

A: 不推荐。用一个容量为 1 的 channel 可以实现互斥：ch <- struct{}{} 获取锁，<-ch 释放锁。但由于 channel 操作涉及运行时锁和可能的 goroutine 切换，性能远低于 sync.Mutex。只有在需要把锁和通信语义结合在一起的特殊场景下才考虑使用 channel 做锁。

Q: for range 遍历 channel 时，如果 channel 没有关闭会怎样？

A: 会永久阻塞当前 goroutine。for range 会持续从 channel 接收数据，直到 channel 被关闭且所有数据被读取完毕。如果 channel 永远不会关闭且没有数据发送，当前 goroutine 将会永远挂起在 range 上。在退出程序前，务必确保 channel 的发送方调用了 close。

Q: 如何让 select 在没有任何 case 就绪时也不阻塞？

A: 给 select 加上 default 分支即可。当没有任何 case 可以执行时，select 会立即执行 default 而不阻塞。这种方法常用于实现非阻塞 send/receive、心跳检测和 polling 模式。

Q: channel 传递大对象时有什么性能建议？

A: channel 发送的是值的副本。如果传递的是大结构体，会触发一次完整的内存拷贝，影响性能。建议传递指针或切片引用。但要注意：传指针意味着接收方可能修改发送方的数据，需要提前评估并发安全性。

最佳实践

1. Channel 所有权管理

定义清晰的规范：谁创建谁就负责关闭，接收方只读取不关闭。

2. 优先使用有缓冲 channel

有缓冲 channel 可以减少发送方的阻塞等待时间，提高并发吞吐量。缓冲大小应该根据生产者-消费者速率差来设计。

3. 用 select + default 实现非阻塞操作

select {
case ch <- value:
    // 操作成功
default:
    // 操作会被阻塞，跳过
}

4. 用 for-range 自动检测 channel 关闭

for value := range ch {
    // channel 关闭后自动退出循环
}

探索更多技术文章

浏览归档，发现更多关于系统设计、工具链和工程实践的内容。

全部文章返回首页