第11章 Channel(重点)
引言:Channel 是 Go 并发模型的灵魂。它源自 Hoare 提出的 CSP(Communicating Sequential Processes)思想——“不要通过共享内存来通信,而要通过通信来共享内存”。本章从
hchan的运行时结构出发,逐层剖析有缓冲/无缓冲 channel 的收发流程、close与range的语义,并落到工程中的最佳实践与常见坑。读懂本章,是理解下一章 select 的基础。
Channel 原理
1. 是什么
Channel 是 Go 语言提供的一等公民(first-class citizen)类型,用于在 goroutine 之间传递数据与同步。它的字面量语法是 chan T,可以通过 make 创建:
ch := make(chan int, 3) // 有缓冲,容量 3
ch := make(chan int) // 无缓冲
Channel 支持三个核心操作:发送 ch <- v、接收 v := <-ch、关闭 close(ch)。它本身是并发安全的——多个 goroutine 可以同时对同一个 channel 收发,不会产生数据竞争,无需额外的锁。
2. 为什么这样设计:CSP 模型与底层数据结构
Go 的并发哲学写在 Effective Go 的开篇:
Do not communicate by sharing memory; instead, share memory by communicating.
这句话对应两种并发风格:
| 风格 | 代表语言 | 同步手段 | Go 中的体现 |
|---|---|---|---|
| 共享内存 + 锁 | C/C++/Java | mutex、condition variable | sync.Mutex、sync.WaitGroup |
| 消息传递 | Erlang/Go | channel、actor mailbox | chan、select |
Go 不是非此即彼,而是以 channel 为主、锁为辅。channel 把“传递数据“和“同步“合二为一:一次 <- 操作既传值又隐含 happened-before 关系,比手工加锁更难写错。
底层实现上,每个 channel 都是一个 hchan 结构体(位于 runtime/chan.go),它包含:
- 一个互斥锁
lock,保证并发安全; - 一个环形缓冲区
buf(无缓冲时为空); - 两个等待队列
sendq/recvq,分别保存因发送/接收而阻塞的 goroutine(用sudog包装); - 元素类型、元素大小、关闭标志等元信息。
所有的 <- 操作都走 runtime 函数:runtime.chansend1 / runtime.chanrecv1,它们会先加 hchan.lock,再根据缓冲区状态决定是直接完成、还是把当前 goroutine 挂起(gopark)。被挂起的 goroutine 通过 sudog 挂到等待队列上,等对端操作时由 goready 唤醒。
整个收发流程可以简化为下图:
┌──────────────────── hchan ─────────────────────┐
│ lock │
│ ┌──── buf (ring buffer) ──────┐ │
│ │ [0][1][2] ... dataqsiz-1 │ │
│ │ ↑recvx ↑sendx │ │
│ └─────────────────────────────┘ │
│ qcount / dataqsiz / closed │
│ recvq ──► sudog ─► sudog ─► nil (等待接收) │
│ sendq ──► sudog ─► sudog ─► nil (等待发送) │
└────────────────────────────────────────────────┘
send 路径: chansend -> lock -> [buf 未满? 写 buf : 入 sendq & gopark] -> unlock
recv 路径: chanrecv -> lock -> [buf 非空? 读 buf : sendq 有? 直接拿 : 入 recvq & gopark] -> unlock
3. 工程实践与常见坑
何时用 channel,何时用锁?
- 数据在 goroutine 间流动、传递所有权 → 用 channel(pipeline、fan-out、结果汇总)。
- 保护一段共享状态的临界区(如缓存、计数器) → 用
sync.Mutex。 - 信号通知、done/cancel → channel(
close(ch)广播)或context.Context。
常见坑:
- goroutine 泄漏:向一个无人接收的无缓冲 channel 发送,或向已满 channel 发送而无人接收,goroutine 永久阻塞。建议配合
context或带超时的select。 - 向已关闭 channel 发送 → panic:
send on closed channel。关闭责任应交给唯一的发送方。 - 重复关闭 → panic:
close of closed channel。可用sync.Once或方向受限 channel 规避。 - nil channel 的妙用:向 nil channel 收发会永久阻塞,在
select中用 nil case 可“动态禁用“某个分支(见第 12 章)。
经验法则:channel 的所有者(创建者 / 唯一发送方)负责关闭;接收方永远不要关闭。这条规则能消除 90% 的 channel 关闭 panic。
hchan
1. 是什么
hchan 是 channel 在运行时的“真身“。你写的 chan int 在编译期是一个 *hchan 指针——make(chan int, n) 实际上调用了 runtime.makechan,分配并初始化一个 hchan 结构。所有 <- 操作最终都转化为对这块内存的读写。
2. 底层数据结构(Go 1.21+,runtime/chan.go)
下面是简化后的关键结构(省略了 GC 相关字段):
type hchan struct {
qcount uint // 当前 buf 中元素个数
dataqsiz uint // buf 的容量(环形数组长度)
buf unsafe.Pointer // 指向环形数组首元素
elemsize uint16 // 单个元素大小(字节)
closed uint32 // 是否已关闭,0=未关闭,1=已关闭
elemtype *_type // 元素类型指针
sendx uint // 下一次发送写入 buf 的下标
recvx uint // 下一次接收读取 buf 的下标
recvq waitq // 等待接收的 sudog 队列
sendq waitq // 等待发送的 sudog 队列
lock mutex // 保护上述所有字段的互斥锁
}
// waitq 是一个双向链表
type waitq struct {
first *sudog
last *sudog
}
// sudog 是 goroutine 在等待队列中的"票据",包装了 g 和数据地址
type sudog struct {
g *g // 被阻塞的 goroutine
next *sudog // 链表后继
prev *sudog // 链表前驱
elem unsafe.Pointer // 数据地址(发送:源;接收:目标)
isSelect bool // 是否处于 select 场景
success bool // 唤醒后是否成功完成操作
c *hchan // 所属 channel
// ... 其余字段用于 GC 与 debug
}
逐字段解释:
| 字段 | 作用 |
|---|---|
qcount | buf 里现在有几个元素。len(ch) 直接返回它。 |
dataqsiz | buf 容量。cap(ch) 返回它。make(chan T, n) 的 n 即此处。无缓冲 channel 为 0。 |
buf | 指向 dataqsiz * elemsize 大小的数组。无缓冲时为 nil。 |
elemsize | 元素字节数,用于 memmove 拷贝。chan struct{} 时为 0,零拷贝。 |
closed | 关闭标志。原子读,但写入受 lock 保护。 |
elemtype | 元素类型,用于拷贝时的边界检查与 GC 扫描。 |
sendx / recvx | 环形缓冲的写/读游标,每次操作后对 dataqsiz 取模前进。 |
recvq / sendq | 因“接收不到“或“发不出去“而阻塞的 goroutine 链表,FIFO。 |
lock | 自旋锁(runtime.mutex),保护所有字段。channel 慢就慢在这把锁——高频收发会成为瓶颈。 |
sudog 的关键字段:
g:被挂起的 goroutine 指针,goready(s.g)用来唤醒它。elem:数据缓冲地址。发送时指向待发送变量;接收时指向接收变量。无缓冲 channel 直接在两个 goroutine 的栈之间通过elem做memmove,绕过 buf。isSelect:标识该 sudog 是否参与select。select 唤醒时需要让“未中标“的 channel 把 sudog 从队列摘除。success:唤醒后用以区分是“真正完成收发“还是“被 select 的另一个分支抢先“。
内存布局示意:
make(chan int, 3) 产生:
ch ──► ┌──────── hchan ────────────┐
│ qcount=0 dataqsiz=3 │
│ elemsize=8 elemtype=int │
│ closed=0 │
│ sendx=0 recvx=0 │
│ recvq={nil,nil} │
│ sendq={nil,nil} │
│ lock ──┐ │
└────────┼──────────────────┘
│
buf ────────────┴──► [ 0 ][ 0 ][ 0 ] (3 个 int 槽位)
^sendx,recvx 都从 0 开始
3. 工程实践与常见坑
chan struct{}是零成本信号:elemsize=0,buf不分配,memmove跳过。适合做“事件通知 / done channel“。len(ch)/cap(ch)是 O(1):直接读qcount/dataqsiz,但只是瞬时快照,不要用它做同步判断(如if len(ch) > 0 { <-ch }仍可能有竞争,应直接<-ch或用select+default)。- 大 channel 是性能杀手:
dataqsiz很大时一次性分配大块内存。若用 channel 做任务队列,考虑用切片 +sync.Cond或第三方 ring buffer。 - channel 不是免费的锁:每次收发要加
hchan.lock+ 可能的memmove+ 可能的 goroutine 调度。高频路径上,sync.Mutex保护一个 slice 通常更快。
有缓冲 Channel
1. 是什么
有缓冲 channel 在创建时指定容量 n > 0:
ch := make(chan int, 3)
它的语义是异步的:发送方在 buf 未满时不阻塞,直接把值丢进 buf 就返回;接收方在 buf 非空时也能立即取走。只有当 buf 满了发送方才阻塞,buf 空了接收方才阻塞。
2. 底层实现:环形缓冲区
有缓冲 channel 的核心是 buf 指向的环形数组,配合 sendx / recvx 两个游标:
dataqsiz = 5, qcount = 3 (buf 中有 3 个元素)
buf: [ A ][ B ][ C ][ . ][ . ]
^ ^
recvx=0 sendx=3
(下次从这里读) (下次从这里写)
接收一次: 读 buf[recvx]=A, recvx=(0+1)%5=1, qcount=2
buf: [ A ][ B ][ C ][ . ][ . ] (A 仍在内存但已"出队")
^recvx=1
发送一次 D: 写 buf[sendx]=D, sendx=(3+1)%5=4, qcount=3
buf: [ A ][ B ][ C ][ D ][ . ]
^recvx=1 ^sendx=4
再发送 E: 写 buf[sendx]=E, sendx=(4+1)%5=0, qcount=4
buf: [ E ][ B ][ C ][ D ][ . ] ← sendx 绕回头部
^sendx=0 ^recvx=1
chansend 的核心逻辑(简化伪代码):
func chansend(c *hchan, ep unsafe.Pointer, block bool) bool {
lock(&c.lock)
if c.closed != 0 {
unlock(&c.lock)
panic("send on closed channel")
}
// 1. 优先:有接收者在等 → 直接把数据拷给接收者,绕过 buf
if sg := c.recvq.dequeue(); sg != nil {
sendDirect(c.elemtype, sg, ep)
unlock(&c.lock)
goready(sg.g) // 唤醒接收者
return true
}
// 2. buf 未满 → 写 buf
if c.qcount < c.dataqsiz {
qp := chanbuf(c, c.sendx)
typedmemmove(c.elemtype, qp, ep)
c.sendx++
if c.sendx == c.dataqsiz { c.sendx = 0 }
c.qcount++
unlock(&c.lock)
return true
}
// 3. buf 满了 → 非阻塞模式直接返回 false;阻塞模式入 sendq & gopark
if !block {
unlock(&c.lock)
return false
}
gp := getg()
mysg := acquireSudog()
mysg.g = gp
mysg.elem = ep
mysg.c = c
c.sendq.enqueue(mysg)
gopark(chanparkcommit, ...) // 当前 goroutine 挂起
// 被唤醒后从这里继续
releaseSudog(mysg)
return true
}
注意第 1 步的优化:即使 buf 有空间,只要
recvq上有人在等,就跳过 buf 直接把数据递到接收者手上。这避免了“先写 buf 再从 buf 读“的双重拷贝。
chanrecv 的对称逻辑:
func chanrecv(c *hchan, ep unsafe.Pointer, block bool) (received bool) {
lock(&c.lock)
// 0. channel 已关闭且 buf 空 → 返回零值
if c.closed != 0 && c.qcount == 0 {
unlock(&c.lock)
if ep != nil { typedmemclr(c.elemtype, ep) }
return false
}
// 1. 有发送者在等 → 直接从发送者那里拿(无缓冲)或从 buf 拿并让发送者补位
if sg := c.sendq.dequeue(); sg != nil {
recv(c, sg, ep)
unlock(&c.lock)
goready(sg.g)
return true
}
// 2. buf 非空 → 读 buf
if c.qcount > 0 {
qp := chanbuf(c, c.recvx)
typedmemmove(c.elemtype, ep, qp)
c.recvx++
if c.recvx == c.dataqsiz { c.recvx = 0 }
c.qcount--
unlock(&c.lock)
return true
}
// 3. buf 空 → 入 recvq & gopark
// ... 类似 chansend
}
3. 工程实践与常见坑
- 缓冲大小是工程权衡:太小 → 高频场景容易阻塞;太大 → 内存占用高、且会“延迟“对背压的感知。常见经验值是
1或2,真正需要削峰填谷时再加大。 - 有缓冲 ≠ 解耦:很多人以为“加了缓冲就不会阻塞“,错。buf 一旦满了发送方照样阻塞。要做真正的解耦,配合
select+default或context主动放弃。 - FIFO 但不保证强实时:channel 保证元素进入 buf 的顺序 = 离开 buf 的顺序,但不保证“发送返回“和“接收方处理完“的时序——这是异步语义。
- 不要用
len(ch) == cap(ch)做判断:仍是瞬时快照,多 goroutine 下不可靠。
// 反面教材
if len(ch) < cap(ch) {
ch <- v // 仍可能阻塞:别的 goroutine 抢先塞进来了
}
// 正确做法
select {
case ch <- v:
default:
// 队列满,降级处理
}
无缓冲 Channel
1. 是什么
无缓冲 channel 创建时不带容量:
ch := make(chan int) // 等价于 make(chan int, 0)
它的语义是同步 rendezvous(会合):发送方和接收方必须“同时在场“才能完成这次传递。一次 ch <- v 在有接收方准备好之前绝不返回;一次 <-ch 在有发送方准备好之前也绝不返回。可以理解为一次同步握手。
2. 底层实现:直接 goroutine-to-goroutine 拷贝
无缓冲 channel 的 dataqsiz=0,buf=nil,所以永远不经过 buf。数据直接从一个 goroutine 的栈拷贝到另一个 goroutine 的栈。
发送路径(无接收者时):
goroutine A: ch <- 42
┌────────────┐ ┌────────────┐
│ g:A │ buf=nil │ recvq=nil │
│ elem=&v │ │ sendq=[] │
└────────────┘ └────────────┘
1. lock; 2. recvq 空、buf 空 → 创建 sudog{g:A, elem:&v}
3. sendq.enqueue(sudog); 4. gopark(A) ← A 挂起
接收方出现时:
goroutine B: x := <-ch
┌────────────┐ 1. lock ┌────────────┐
│ g:B │ 2. 发现 sendq 有 A 的 sudog │ sendq=[A] │
│ elem=&x │ 3. memmove(x, A.elem) ← 直接拷贝 │ │
└────────────┘ 4. goready(A) 唤醒 A └────────────┘
5. B 直接返回,无需 gopark
关键点:
- 数据
42从 A 的栈变量v直接 memmove 到 B 的栈变量x,没有中间 buf。 - B 拿到锁后,A 已经
gopark,但 A 的sudog.elem=&v仍指向 A 栈上的v——只要 A 没被唤醒、栈没销毁,这个地址就有效。这正是gopark的作用:冻结 goroutine 状态。 - B 唤醒 A 后,A 从
chansend中gopark之后的那行继续执行,释放 sudog 并返回。
happened-before 关系:
无缓冲 channel 建立强同步:ch <- v 的“完成“在 <-ch 的“完成“之前。因此 v 的写入对接收方完全可见,无需额外同步。
A: v = 42; ch <- v; // A 写 v 在 send 之前
B: x := <-ch; print(x); // B 接收在读 x 之前,且能看到 A 对 v 的写
3. 工程实践与常见坑
- 无缓冲 channel = 同步原语:常用于“等对方完成“的握手,如:
done := make(chan struct{})
go func() {
// 工作...
done <- struct{}{} // 等主 goroutine 准备好接收
}()
<-done // 阻塞直到 worker 完成
- 主 goroutine 直接
go f()后立刻<-ch会卡住 worker:如果主 goroutine 还没准备好接收,worker 在ch <- v处阻塞,看似“无限快“的 worker 也跑不起来。 chan struct{}是最优无缓冲信号:零拷贝、零内存,纯同步语义。- 不要把无缓冲 channel 当队列用:它没有“暂存“能力,任何一方先到都得等。需要暂存就用有缓冲。
- deadlock 经典坑:
func main() {
ch := make(chan int)
ch <- 1 // 主 goroutine 阻塞,没人接收 → fatal error: all goroutines are asleep - deadlock!
fmt.Println(<-ch)
}
修复:先 go func() { <-ch }(),或改成 make(chan int, 1)。
close()
1. 是什么
close(ch) 把 channel 标记为“不再有数据发送“。它有两个直接后果:
- 后续的所有接收会立即返回:先把 buf 里剩余元素按 FIFO 消费完,之后返回零值,且
v, ok := <-ch的ok为false。 - 后续的任何发送都会 panic:
send on closed channel。
ch := make(chan int, 2)
ch <- 1
ch <- 2
close(ch)
fmt.Println(<-ch) // 1
fmt.Println(<-ch) // 2
v, ok := <-ch // v=0, ok=false ← buf 空了,返回零值
2. 底层实现:runtime.closechan
func closechan(c *hchan) {
if c == nil {
panic("close of nil channel") // nil channel 不能 close
}
lock(&c.lock)
if c.closed != 0 {
unlock(&c.lock)
panic("close of closed channel") // 重复 close panic
}
c.closed = 1 // 置位关闭标志
var glist gList
// 1. 唤醒所有接收等待者:他们都会得到零值 + ok=false
for {
sg := c.recvq.dequeue()
if sg == nil { break }
sg.elem = nil // 标记:收到的是零值
gp := sg.g
gp.param = nil
glist.push(gp)
}
// 2. 唤醒所有发送等待者:他们会被 panic
for {
sg := c.sendq.dequeue()
if sg == nil { break }
sg.elem = nil
gp := sg.g
gp.param = nil
glist.push(gp)
}
unlock(&c.lock)
// 3. 统一 goready 所有挂起的 goroutine
for !glist.empty() {
gp := glist.pop()
gp.schedlink = 0
goready(gp) // 发送者唤醒后会 panic
}
}
关键设计:
- 置
closed=1在锁内:与chansend的c.closed != 0检查互斥,避免 close 与 send 竞争。 - 批量唤醒:所有等待者先收集到
glist,释放锁后再goready,缩短锁持有时间。 - 发送等待者也会被唤醒:但它们的
chansend在gopark返回后会发现c.closed != 0,于是 panic——这就是“向已关闭 channel 发送会 panic“的运行时根源。 - 接收者得到零值:
close时如果recvq上有人,他们的sudog.elem被置 nil,唤醒后chanrecv走零值路径。
3. 工程实践与常见坑
三大 panic 场景:
| 操作 | 条件 | 结果 |
|---|---|---|
close(ch) | ch 已关闭 | close of closed channel |
close(ch) | ch 是 nil | close of nil channel |
ch <- v | ch 已关闭 | send on closed channel |
关闭责任与安全关闭模式:
原则:只有发送方关闭,且只关闭一次。具体落地有三种模式:
模式 1:方向限制(推荐)
package main
import "fmt"
// producer 只拿到发送方向,外部拿到接收方向
func producer(out chan<- int) {
defer close(out) // 唯一发送方负责关闭
for i := 0; i < 3; i++ {
out <- i
}
}
func main() {
ch := make(chan int)
go producer(ch)
for v := range ch { // 接收方用 range 自动检测关闭
fmt.Println(v)
}
}
chan<- int 让 producer 内部无法接收、外部无法发送,关闭责任唯一明确。
模式 2:sync.Once
var once sync.Once
ch := make(chan int)
func shutdown() {
once.Do(func() { close(ch) })
}
适合多个 goroutine 都可能触发关闭的场景。
模式 3:额外的 done channel / context
不直接 close 数据 channel,而是用一个独立的 done 信号通知所有发送方停止发送:
ctx, cancel := context.WithCancel(context.Background())
go func() {
for {
select {
case <-ctx.Done():
return
case ch <- produce():
}
}
}()
// 取消时调用 cancel(),发送方自行退出,再由所有者关闭 ch
关闭 nil channel 的妙用:在 select 中把一个 channel 变量置 nil,对应 case 会永久阻塞(即“禁用“该分支),常用于“处理完一类事件后不再处理“的状态机。详见第 12 章。
range
1. 是什么
for v := range ch 是遍历 channel 的语法糖。它会不断接收直到 channel 被关闭且 buf 排空才退出循环:
for v := range ch {
fmt.Println(v)
}
// 等价于
for {
v, ok := <-ch
if !ok { // channel 关闭且 buf 空
break
}
fmt.Println(v)
}
2. 底层实现
for range chan 在编译期被改写为调用 runtime.chanrecv2(带 ok 返回值的版本)。每次迭代:
- 调用
chanrecv,传入接收变量地址; - 若
chanrecv返回false(channel 已关闭且 buf 空)→ 跳出循环; - 若返回
true→ 执行循环体,回到步骤 1。
注意 chanrecv 在 channel 关闭后会先消费完 buf 里的剩余元素,每消费一个返回 true,buf 空了才返回 false。所以 range 不会丢数据。
channel 状态: closed=1, buf=[A, B, C]
range 第 1 次: chanrecv -> 读 A, 返回 true → 循环体
range 第 2 次: chanrecv -> 读 B, 返回 true → 循环体
range 第 3 次: chanrecv -> 读 C, 返回 true → 循环体
range 第 4 次: chanrecv -> closed && buf 空 -> 返回 false → break
3. 工程实践与常见坑
- 必须有发送方关闭:
for range ch在 channel 永不关闭时会永久阻塞最后一条chanrecv,导致 goroutine 泄漏。生产者完成后必须close(ch)。 - 不要在接收方 close:
range的接收方不知道发送方何时停,强求关闭会引入竞争。 - range 不会消费 nil channel:
for v := range nilCh永久阻塞(与<-nil一致)。 - break 只跳出 range:在
select内的break跳不出外层for range,需要标签:
outer:
for v := range ch {
select {
case <-stop:
break outer // 用标签跳出外层
default:
process(v)
}
}
- range 一个有缓冲 channel 时关闭后仍能读出残留数据:这是特性不是 bug,确保不丢消息。但要小心:如果发送方在
close前 buf 里还有 N 条未消费,接收方的range会先消费这 N 条再退出。
select
1. 是什么
select 是 Go 中处理多个 channel 操作的控制结构,类似 switch,但每个 case 必须是 channel 的发送或接收:
select {
case v := <-ch1:
fmt.Println("from ch1:", v)
case ch2 <- 42:
fmt.Println("sent to ch2")
case <-time.After(time.Second):
fmt.Println("timeout")
default:
fmt.Println("no channel ready")
}
它的语义:
- 阻塞直到至少一个 case 就绪(除非有
default); - 若多个 case 同时就绪,随机选一个执行;
- 每个 case 的操作与对应分支体原子地完成(其他 case 不会被同时执行)。
2. 为什么这样设计
select 是 CSP 模型中“非确定性选择“的实现,它让程序能公平地等待多个事件源,而不是轮询。底层由 runtime.selectgo 实现,涉及 scase 数组、对所有 channel 加锁、随机洗牌等机制——细节留到第 12 章详述。
这里只需理解一个高层流程:
┌─────────────┐
│ select { } │
└──────┬──────┘
│
┌─────────────┼─────────────┐
▼ ▼ ▼
case <-ch1 case ch2<-v case <-ch3
│ │ │
└─────────────┼─────────────┘
▼
1. 随机洗牌 case 顺序
2. 依次尝试每个 case 是否就绪
3. 任一就绪 → 执行该 case,返回
4. 全部未就绪且有 default → 执行 default
5. 全部未就绪且无 default → 在所有 channel 上
挂起 sudog,gopark,等待任一 channel 唤醒
│
▼
被唤醒 → 清理其它 channel 上的 sudog → 执行中奖 case
3. 工程实践与常见坑
- 超时控制:
case <-time.After(d)是最常用的模式,避免 goroutine 永久阻塞。 - 非阻塞收发:
default分支让 select 立即返回,常用于“有就处理、没有就跳过“。 - 动态禁用 case:把 channel 变量置
nil,对应 case 永久阻塞,等价于“从 select 中移除“。 - 空 select
select{}:永久阻塞,常用于让 main goroutine 等待信号(避免泄漏的 goroutine 退出前 main 退出)。
select的完整原理(scase结构、selectgo算法、随机性与公平性)见 第 12 章 select。
Channel 最佳实践
1. 所有权与关闭责任
原则:channel 的创建者即唯一发送者,由它负责 close;接收者永远不要 close。
落地方式:用方向受限 channel 显式表达所有权:
package main
import "fmt"
// 返回只读 channel,调用方只能接收;内部 goroutine 拥有写端并负责关闭
func counter() <-chan int {
out := make(chan int)
go func() {
defer close(out)
for i := 0; i < 5; i++ {
out <- i
}
}()
return out
}
func main() {
for v := range counter() {
fmt.Println(v)
}
}
这样编译器会阻止接收方误发或误关,把“关闭责任“从约定升级为类型约束。
2. Pipeline 模式
把多个 stage 用 channel 串起来,每个 stage 是一组 goroutine:
package main
import "fmt"
func gen(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 n := range in {
out <- n * n
}
}()
return out
}
func main() {
for v := range square(gen(1, 2, 3, 4)) {
fmt.Println(v) // 1 4 9 16
}
}
每个 stage 输入 <-chan int、输出 <-chan int,方向受限、关闭责任清晰,可自由组合。
3. Fan-out / Fan-in
Fan-out:多个 worker 消费同一个 channel,并行处理。Fan-in:把多个 channel 的结果汇入一个。
package main
import (
"fmt"
"sync"
)
func worker(id int, in <-chan int, out chan<- int, wg *sync.WaitGroup) {
defer wg.Done()
for n := range in {
out <- n * n
}
}
func merge(cs ...<-chan int) <-chan int {
var wg sync.WaitGroup
out := make(chan int)
output := func(c <-chan int) {
defer wg.Done()
for v := range c {
out <- v
}
}
for _, c := range cs {
wg.Add(1)
go output(c)
}
go func() {
wg.Wait()
close(out) // 所有输入消费完才关闭合并 channel
}()
return out
}
func main() {
in := make(chan int)
out := make(chan int)
var wg sync.WaitGroup
// Fan-out: 3 个 worker
for i := 0; i < 3; i++ {
wg.Add(1)
go worker(i, in, out, &wg)
}
go func() {
for i := 0; i < 10; i++ {
in <- i
}
close(in)
}()
// 等所有 worker 退出后关闭 out
go func() { wg.Wait(); close(out) }()
for v := range out {
fmt.Println(v)
}
}
4. 超时与取消(done channel / context)
package main
import (
"context"
"fmt"
"time"
)
func worker(ctx context.Context) <-chan int {
out := make(chan int)
go func() {
defer close(out)
for i := 0; ; i++ {
select {
case <-ctx.Done():
return
case out <- i:
time.Sleep(200 * time.Millisecond)
}
}
}()
return out
}
func main() {
ctx, cancel := context.WithTimeout(context.Background(), time.Second)
defer cancel()
for v := range worker(ctx) {
fmt.Println(v)
}
}
ctx.Done() 本质是一个 <-chan struct{},cancel() / 超时会 close 它,所有 select 立刻就绪。
5. Worker Pool
package main
import (
"fmt"
"sync"
)
func main() {
jobs := make(chan int, 100)
results := make(chan int, 100)
// 启动 4 个 worker
var wg sync.WaitGroup
for w := 0; w < 4; w++ {
wg.Add(1)
go func() {
defer wg.Done()
for j := range jobs {
results <- j * j
}
}()
}
// 投递任务
go func() {
for i := 0; i < 20; i++ {
jobs <- i
}
close(jobs)
}()
// 等所有 worker 退出后关闭 results
go func() { wg.Wait(); close(results) }()
for r := range results {
fmt.Println(r)
}
}
6. 常见反模式
反模式 1:用 channel 当互斥锁
// 别这样
ch := make(chan struct{}, 1)
ch <- struct{}{} // "加锁"
// 临界区
<-ch // "解锁"
channel 比 sync.Mutex 慢一个数量级,且语义模糊。需要锁就用 sync.Mutex。
反模式 2:把 channel 当普通集合
// 别这样:用 channel 存数据再反复 len/遍历
ch := make(chan int, 1000)
// ... 塞一堆数据
// 想随机访问?做不到。
channel 是“流“不是“集合“,需要切片/映射就用 slice/map。
反模式 3:发送方未关闭导致 range 泄漏
// 反面
go func() {
ch <- 1
ch <- 2
// 忘了 close(ch)
}()
for v := range ch { // 永久阻塞在第 3 次接收
fmt.Println(v)
}
反模式 4:在接收方 close
// 反面
go func() {
for v := range ch {
if v == sentinel {
close(ch) // 接收方关闭,可能和发送方竞争 → panic
}
}
}()
用 context 或额外的 done channel 通知发送方停止,而不是接收方去 close。
反模式 5:无缓冲 channel 当缓冲用
// 反面:以为这样能并发处理 5 个
ch := make(chan int) // 无缓冲!
for i := 0; i < 5; i++ {
go func() { ch <- work() }() // 实际仍串行:必须有人立即接收
}
需要并发暂存就用 make(chan int, n)。
7. 性能要点
| 关注点 | 建议 |
|---|---|
| 高频收发 | 优先 sync.Mutex + slice;channel 每次有锁+可能调度 |
| 信号通知 | chan struct{},零拷贝零分配 |
| 缓冲大小 | 默认 1;削峰填谷再加大;过大掩盖背压问题 |
| goroutine 泄漏 | 所有发送路径配 ctx.Done() 或超时;用 goleak 工具检测 |
| 批量传递 | 一次发 []T 而非多次发 T,减少锁与调度次数 |
本章小结
本章从 hchan 结构出发,剖析了 channel 的核心实现:
hchan由互斥锁lock、环形缓冲buf、sendq/recvq等待队列构成;len/cap是 O(1) 快照。- 有缓冲 channel 用环形数组 +
sendx/recvx游标实现 FIFO 队列;满则发送方入sendq,空则接收方入recvq。一个重要优化:只要recvq有等待者,发送方会绕过 buf 直接把数据递给接收者。 - 无缓冲 channel 同步会合,
buf=nil,数据在两个 goroutine 栈间直接memmove,建立强 happened-before 关系。 close()在锁内置位closed,批量唤醒recvq(得零值)和sendq(触发 panic)。三大 panic:重复关闭、关闭 nil、向已关闭 channel 发送。range是chanrecv2的语法糖,消费完 buf 中残留后才退出。select是多路 channel 复用,随机选择就绪分支,底层selectgo详见下一章。- 最佳实践:方向受限表达所有权、发送方负责关闭、pipeline/fan-out/context 配合、避免把 channel 当锁或集合。
掌握 channel 的关键在于理解它“传递数据 + 同步“二合一的本质,以及“所有者关闭、接收方只读“的所有权模型。下一章 select 会深入 scase 与 selectgo,揭示随机性与公平性的运行时实现。