Blog

Go 的通道与 select

作者

Go 通道在 goroutine 之间传递值，也让它们互相等待。本文讲无缓冲和有缓冲通道、关闭、死锁、nil 通道，以及 select 如何同时等待多个通道。

单靠一个 goroutine，没法把结果交回来。它没有能让你接住的返回值。Go 给出的答案是通道（channel）：一根带类型的管子，一个 goroutine 往里发送，另一个从里面接收。

本文讲怎样创建通道、无缓冲和有缓冲通道的区别、关闭通道、你迟早会碰到的死锁错误，以及用 select 同时等待多个通道。下面每个程序都在 Go 1.26 上跑过，输出直接从运行结果粘贴而来。

创建通道，发送和接收

Go 的通道用 make 创建，类型是 chan T，其中 T 是它传递的值的类型。箭头运算符 <- 两件事都干：ch <- v 发送 v，<-ch 接收一个值。

package main

import "fmt"

func main() {
	ch := make(chan string)

	go func() {
		ch <- "hello from a goroutine"
	}()

	msg := <-ch
	fmt.Println(msg)
	fmt.Printf("%T\n", ch)
}

输出：

hello from a goroutine
chan string

goroutine 发送一个字符串，main 接收它。这里没有 sleep，也没有循环去检查 goroutine 是否结束。<-ch 会一直等到有值到达。这种等待，是通道的另一半作用。

只发送和只接收的通道类型

函数参数可以承诺只朝一个方向使用通道。chan<- int 是只能发送的通道，<-chan int 是只能接收的通道。箭头相对 chan 的位置，表示值流动的方向。

package main

import "fmt"

func produce(out chan<- int) {
	for i := range 3 {
		out <- i * i
	}
	close(out)
}

func consume(in <-chan int) int {
	total := 0
	for v := range in {
		total += v
	}
	return total
}

func main() {
	ch := make(chan int)
	go produce(ch)
	fmt.Println(consume(ch))
}

输出：

5

main 给两个函数传的都是普通的 chan int，Go 在每次调用时把它转换成更窄的类型。produce 发送 0、1 和 4，consume 把它们加起来。close 和 for range 在下面有专门的一节。

更窄的类型由编译器检查。如果 consume 试图发送，构建就会停下：

package main

func consume(in <-chan int) {
	in <- 1
}

func main() {
	ch := make(chan int)
	go consume(ch)
	<-ch
}

构建失败，报错：

./main.go:4:2: invalid operation: cannot send to receive-only channel <-chan int in (variable of type <-chan int)

这条消息把类型写了两遍，读起来有点怪，但意思很清楚。方向类型在运行时没有任何开销，又由编译器强制执行，所以只要函数只发送或只接收，就用上它。

无缓冲通道和有缓冲通道

用 make(chan int) 创建的通道是无缓冲的：它没有地方存值。用 make(chan int, 2) 创建的通道是有缓冲的，能存两个值。就这一点区别，决定了发送什么时候必须等待。

  • 在无缓冲通道上，发送要等到另一个 goroutine 接收，接收要等到另一个 goroutine 发送。两边碰头，值传过去，然后各自继续。这种碰头常被称为会合（rendezvous）。
  • 在有缓冲通道上，只有缓冲区满了发送才等待，只有缓冲区空了接收才等待。

先看两种通道的动画，再读背后的程序：

无缓冲：make(chan int) 发送方 窗口 接收方 1 阻塞：还没有接收方 交接完成：双方继续 有缓冲：make(chan int, 2) 发送方 队头 len 0, cap 2 len 1, cap 2 len 2, cap 2 接收方 1 2 3 阻塞：架子已满 无缓冲：发送方递出 1，没人接收，只能等 接收方到了，从窗口接过 1，双方继续执行 有缓冲，cap 2：发送 1 和 2 直接放上架子，不用等 两个格子都满了，所以发送 3 只能等 接收方从队头取走 1，2 前移，空出一格 3 放进空格，发送方继续执行

上方是无缓冲通道，下方是容量为 2 的有缓冲通道。无缓冲的发送要等接收方把值拿走。有缓冲时，发送 1 和 2 马上完成，因为还有空格子；发送 3 要等，因为架子满了，等接收方取走 1、空出一格，它就立刻完成。

如果动画没有播放，下面用文字把这几步再说一遍：

  1. 在无缓冲通道上，发送方递出 1。没人接收，所以发送方被阻塞。
  2. 接收方到了，接过 1。发送和接收同时完成，两个 goroutine 都继续执行。
  3. 在容量为 2 的有缓冲通道上，发送方先发送 1，再发送 2。两个值都放进了空格子，所以哪次发送都不用等。此时 len 是 2。
  4. 发送方试着发送 3。两个格子都满了，所以这次发送被阻塞。
  5. 接收方取走队头的值 1。2 前移，空出一格。
  6. 3 放进空格，被阻塞的发送完成，发送方继续执行。

下面的程序做的是同样几步，由一个 goroutine 充当发送方，main 充当接收方。只有 main 打印，而且每次打印都发生在某个必然已经完成的通道操作之后，所以每次运行的输出顺序都不会变：

package main

import "fmt"

func main() {
	// Unbuffered: the hatch has no room, so a send waits for a receiver.
	hatch := make(chan int)
	sent := make(chan bool)
	go func() {
		hatch <- 1 // waits here until main receives
		sent <- true
	}()
	fmt.Println("hatch: received", <-hatch)
	<-sent
	fmt.Println("hatch: the sender has moved on")

	// Buffered: the shelf has 2 slots, so 2 sends finish with nobody receiving.
	shelf := make(chan int, 2)
	ready := make(chan bool)
	done := make(chan bool)
	go func() {
		shelf <- 1
		shelf <- 2
		ready <- true
		shelf <- 3 // the shelf is full, so this waits
		done <- true
	}()
	<-ready
	fmt.Println("shelf: len", len(shelf), "cap", cap(shelf))
	fmt.Println("shelf: received", <-shelf)
	<-done
	fmt.Println("shelf: the send of 3 has finished, len", len(shelf))
	fmt.Println("shelf: received", <-shelf)
	fmt.Println("shelf: received", <-shelf)
}

输出：

hatch: received 1
hatch: the sender has moved on
shelf: len 2 cap 2
shelf: received 1
shelf: the send of 3 has finished, len 2
shelf: received 2
shelf: received 3

前半段里，main 拿走 1 之前，发送方到不了 sent <- true。后半段里，goroutine 执行到 ready <- true 时，还没有人从 shelf 接收，所以前两次发送都没有等。发送 3 要等 main 取走一个值之后才完成。值出来的顺序和进去的顺序一样，因为通道是先进先出的。

cap 是你传给 make 的缓冲区大小，len 是此刻正在等待的值的个数。无缓冲通道的两者都是 0。不过在实际代码里，别根据 len 做决定。等你按它行动时，别的 goroutine 可能已经改变了它。

用十岁孩子能懂的话说

想象一个餐馆后厨，有两个厨师。一个做菜，另一个把菜装上托盘。

无缓冲通道是两人之间墙上的一个小传菜窗口。窗口没有台面。第一个厨师把盘子举到窗口，在第二个厨师从另一边接过去之前，不能松手。如果第二个厨师在忙，第一个厨师就只能端着盘子站在那儿。盘子一换手，两个厨师就都回去干活。

有缓冲通道是一个格子数量固定的架子，比如两个格子。只要有空格子，第一个厨师就可以放下盘子走开。两个格子都满了，厨师就得站着等，直到空出一格。第二个厨师从架子前头拿盘子。如果架子是空的，第二个厨师就等着。

准确的说法

通道是一个内部带锁的队列，由所有持有它的 goroutine 共享。无缓冲通道的队列大小是零。在它上面的发送，只有等接收方直接拿走值才会完成，所以两个 goroutine 必然碰头。无缓冲的发送返回后，你就知道接收方已经拿到了值。

容量为 N 的有缓冲通道最多存放 N 个值。只有已经有 N 个值在等待时，发送才阻塞；只有一个值都没有时，接收才阻塞。有缓冲的发送返回后，你只知道值进了缓冲区。可能还没有人接收它。

“阻塞”的意思是 goroutine 被挂起了。它不会空转，也不占用 CPU。通道能继续推进时，运行时会唤醒它。

这个比喻的局限：厨房的架子能放各种各样的盘子，而通道只传递一种类型。好几个厨师可以在同一个窗口前等，语言并不保证下一个轮到谁。最大的区别在讲关闭的那一节：关上的窗口并不会一直关着，它会永远不停地递出空盘子。

死锁：所有 goroutine 都在等

在无缓冲通道上发送，如果没有别的 goroutine 来接收，就会永远等下去。如果程序里每个 goroutine 都这样卡住，Go 会发现并停止程序：

package main

import "fmt"

func main() {
	ch := make(chan int)
	ch <- 1
	fmt.Println(<-ch)
}

程序打印错误和栈跟踪，然后停止：

fatal error: all goroutines are asleep - deadlock!

goroutine 1 [chan send]:

main 试图发送，而唯一可能接收的 goroutine 就是 main 自己，它现在卡在发送上。下一行的接收永远不会执行。栈跟踪显示了 goroutine 停下时正在做什么：[chan send]。

把第一行改成 make(chan int, 1)，这个程序就能运行，因为值放得进缓冲区。不过在实际代码里，靠加缓冲区让死锁消失，通常只是掩盖了设计问题。

这是 fatal error，不是 panic，所以 recover 捕获不了。而且只有所有 goroutine 都阻塞时，这个检查才会触发。如果一个 goroutine 永远卡在某个通道上，而其他 goroutine 还在运行，比如 HTTP 服务器的那些 goroutine，Go 什么都不会说。这就是 goroutine 泄漏，你只会看到内存不断上涨。

关闭通道

发送方调用 close(ch)，表示“不会再有值了”。接收方仍然可以取走缓冲区里已有的值。取完之后，每次接收都会立刻返回该类型的零值。双返回值的形式告诉你拿到的是哪一种：

package main

import "fmt"

func main() {
	ch := make(chan int, 2)
	ch <- 10
	close(ch)

	v, ok := <-ch
	fmt.Println(v, ok)
	v, ok = <-ch
	fmt.Println(v, ok)
	v, ok = <-ch
	fmt.Println(v, ok)
}

输出：

10 true
0 false
0 false

10 是在关闭之前发送的，所以第一次接收仍然拿到它，ok 为 true。之后通道既关闭又为空，所以后面每次接收都立刻得到 0, false。它不阻塞，也不会停。这就是比喻里说的“永远递出空盘子”。这和在 map 里查键用的 comma-ok 写法是同一个思路。

手动检查 ok 很烦，所以 for v := range ch 替你做了。循环不断接收值，直到通道关闭且为空，然后结束：

package main

import "fmt"

func main() {
	words := make(chan string)

	go func() {
		for _, w := range []string{"flour", "eggs", "milk"} {
			words <- w
		}
		close(words)
	}()

	for w := range words {
		fmt.Println("got", w)
	}
	fmt.Println("the channel is closed, so the loop ended")
}

输出：

got flour
got eggs
got milk
the channel is closed, so the loop ended

如果 goroutine 忘了 close(words)，循环会永远等第四个词，程序最后会以上一节的死锁错误结束。

只有发送方关闭通道

向已关闭的通道发送会 panic，关闭同一个通道两次也会：

package main

import "fmt"

func main() {
	ch := make(chan int, 1)
	close(ch)
	fmt.Println("closed")
	ch <- 1
}

输出第一行，然后停止：

closed
panic: send on closed channel

关闭两次会得到 panic: close of closed channel。接收方无从知道发送方是不是正要发送，所以由接收方关闭通道，就有触发这个 panic 的风险。保证安全的规则很简单：只有发送方关闭通道，而且只在没有更多值要发送时关闭。如果有多个发送方，谁都不该关闭。应该由知道它们全都结束了的那一方来关闭。讲 sync 的那一部分会介绍常用的工具。

你也不必关闭每个通道。通道不是文件。不管有没有关闭，无法再访问的通道都会被垃圾回收器清理。当接收方需要知道值已经发完了的时候，才关闭通道，就像 range 需要的那样。

nil 通道永远阻塞

通道类型的零值是 nil，var ch chan int 得到的就是一个 nil 通道。向 nil 通道发送或从中接收，都会永远阻塞：

package main

import "fmt"

func main() {
	var ch chan int
	fmt.Println(ch == nil)
	<-ch
}

输出 true，然后是死锁错误：

true
fatal error: all goroutines are asleep - deadlock!

goroutine 1 [chan receive (nil chan)]:

栈跟踪里甚至写着 (nil chan)，忘了写 make 时这很有用。关闭 nil 通道也会 panic。

这听起来纯粹是个坑，但它有一个好用处。在 select 里，nil 通道上的 case 永远不会就绪，所以把通道变量设为 nil，就能关掉那个 case。讲合并两个通道的那一节会用到它。

select：同时等待多个通道

select 语句会等到几个通道操作中有一个能进行，然后执行那个 case。它看起来像 switch，但每个 case 都是一次发送或接收：

package main

import "fmt"

func split(nums []int, evens, odds chan<- int, done chan<- bool) {
	for _, n := range nums {
		if n%2 == 0 {
			evens <- n
		} else {
			odds <- n
		}
	}
	done <- true
}

func main() {
	evens := make(chan int)
	odds := make(chan int)
	done := make(chan bool)
	go split([]int{4, 7, 1, 8}, evens, odds, done)

	for {
		select {
		case n := <-evens:
			fmt.Println("even:", n)
		case n := <-odds:
			fmt.Println("odd:", n)
		case <-done:
			fmt.Println("done")
			return
		}
	}
}

输出：

even: 4
odd: 7
odd: 1
even: 8
done

split 把每个数发到两个无缓冲通道中的一个，最后发出 done 信号。每次发送都要等 main 接收，所以同一时刻只有一个 case 就绪，输出保持了输入的顺序。

多个 case 同时就绪时

如果 select 检查时有不止一个 case 就绪，Go 会随机挑一个，每个的机会相等。它不像 switch 那样从上往下试。这是有意为之。如果 select 总是优先第一个 case，一个繁忙的第一个通道就可能让其他通道永远饿死。

上面的程序之所以按固定顺序输出，只是因为它从来不会有两个 case 同时就绪。如果 split 用的是有缓冲通道，几个数可能同时在等待，输出顺序就会在每次运行之间变化。所以别靠 case 的顺序来定优先级。如果某个通道真的必须优先，就先用单独的 select 检查它，或者把流程设计成它根本不用竞争。

default：完全不等

带 default case 的 select 永远不会阻塞。如果此刻没有其他 case 就绪，就执行 default。这样你就得到一种只在能立即完成时才进行的发送或接收：

package main

import "fmt"

func trySend(ch chan<- int, v int) {
	select {
	case ch <- v:
		fmt.Println("sent", v)
	default:
		fmt.Println("full, skipped", v)
	}
}

func tryReceive(ch <-chan int) {
	select {
	case v := <-ch:
		fmt.Println("received", v)
	default:
		fmt.Println("empty, nothing to receive")
	}
}

func main() {
	shelf := make(chan int, 2)
	trySend(shelf, 1)
	trySend(shelf, 2)
	trySend(shelf, 3)
	tryReceive(shelf)
	tryReceive(shelf)
	tryReceive(shelf)
}

输出：

sent 1
sent 2
full, skipped 3
received 1
received 2
empty, nothing to receive

这就是动画里的那个架子，只不过发送 3 不再等待。它放弃了，走了 default。比如你宁可丢掉一条指标数据，也不想拖慢请求，就可以这样用。

在循环里用 default 要小心。套在带 default 的 select 外面的循环永远不会阻塞，所以它会空转，在等待某件事发生的同时烧满一整个 CPU 核心。

用 time.After 设置超时

time.After(d) 返回一个通道，过了 d 之后，这个通道会收到一个值。把它和真正的工作一起放进 select，谁先就绪谁赢：

package main

import (
	"fmt"
	"time"
)

func fetch(delay time.Duration) <-chan string {
	out := make(chan string, 1)
	go func() {
		time.Sleep(delay)
		out <- "report ready"
	}()
	return out
}

func wait(result <-chan string, limit time.Duration) {
	select {
	case r := <-result:
		fmt.Println(r)
	case <-time.After(limit):
		fmt.Println("gave up waiting")
	}
}

func main() {
	wait(fetch(0), time.Second)
	wait(fetch(time.Second), 10*time.Millisecond)
}

输出：

report ready
gave up waiting

第一个任务面对一秒的时限，立刻完成。第二个任务要花一秒，时限却只有 10 毫秒。故意把差距拉得很大，这样在慢机器上结果也一样。

有两个细节值得知道。fetch 用了大小为 1 的缓冲区，所以即使 wait 已经放弃，goroutine 仍然能把迟到的结果发出去，然后退出。如果用无缓冲通道，它会永远阻塞在一个没人接收的发送上。另外，从 Go 1.23 起，time.After 创建的定时器只要不再被引用，即使从没触发过，也会被垃圾回收器清理。以前的建议警告说在循环里用 time.After 会泄漏定时器。现在已经不是这样了。

如果超时需要穿过好几层函数调用，比如一个 HTTP 请求又去调用数据库，就用 context，讲 context 和并发模式的那一部分会介绍。

合并两个通道，用 nil 关掉 case

从两个通道接收，直到两个都关闭，这正是 nil 通道派上用场的地方。一个通道关闭后，把它的变量设为 nil，select 就不会再选那个 case：

package main

import (
	"fmt"
	"slices"
)

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

func main() {
	a := send(1, 2, 3)
	b := send(10, 20)

	var got []int
	for a != nil || b != nil {
		select {
		case n, ok := <-a:
			if !ok {
				a = nil // a nil channel is never ready, so this case is now off
				continue
			}
			got = append(got, n)
		case n, ok := <-b:
			if !ok {
				b = nil
				continue
			}
			got = append(got, n)
		}
	}
	slices.Sort(got)
	fmt.Println(got)
}

输出：

[1 2 3 10 20]

没有 a = nil 的话，关闭后的 a 每轮循环都会就绪，一遍又一遍地给出 0, false，循环就会空转。有了它，这个 case 就再也不会被选中。两个变量都是 nil 时，循环结束。a 和 b 的值到达的顺序每次运行都不一样，所以程序在打印前先排序。

先学两个模式

Go 的很多并发代码，都是由返回通道的小函数搭起来的。有两个现在就值得学。讲 context 和并发模式的那一部分里，工作池（worker pool）和管道都建立在它们之上。

生成器返回只接收通道

生成器是这样一个函数：它启动一个 goroutine，返回一个 <-chan T，并在上面发送值：

package main

import "fmt"

func countdown(from int) <-chan int {
	out := make(chan int)
	go func() {
		defer close(out)
		for i := from; i > 0; i-- {
			out <- i
		}
	}()
	return out
}

func main() {
	for n := range countdown(3) {
		fmt.Println(n)
	}
	fmt.Println("liftoff")
}

输出：

3
2
1
liftoff

返回类型 <-chan int 意味着调用方只能接收，所以只有里面的 goroutine 能发送或关闭。defer close(out) 确保不管 goroutine 怎样结束，通道都会关闭，调用方因此可以用 range。

用 done 通道让 goroutine 停下

永不结束的生成器需要一种办法来接收停止通知，否则它的 goroutine 会永远等在一个没人接收的发送上。常用的信号是一个由调用方关闭的 done 通道：

package main

import "fmt"

func naturals(done <-chan struct{}) <-chan int {
	out := make(chan int)
	go func() {
		defer close(out)
		for i := 1; ; i++ {
			select {
			case out <- i:
			case <-done:
				return
			}
		}
	}()
	return out
}

func main() {
	done := make(chan struct{})
	nums := naturals(done)

	for n := range nums {
		fmt.Println(n)
		if n == 3 {
			break
		}
	}

	close(done)
	for range nums {
		// drain until the generator closes out
	}
	fmt.Println("the generator has stopped")
}

输出：

1
2
3
the generator has stopped

goroutine 同时等两件事：有人接收它的下一个数，或者 done 被关闭。main 跳出循环后，关闭 done。已关闭的通道总是可以接收，所以 goroutine 返回，并关闭 out。第二个循环只有在 out 关闭后才会结束，这证明 goroutine 真的停下了。

这里正确的信号是关闭，而不是发送一个值。一次发送只唤醒一个接收方。一次关闭会唤醒所有接收方，不管有多少 goroutine 在听。元素类型 struct{} 不占内存，它表明值本身不重要，重要的只是这件事发生了。

通过通信来共享内存

Go 的建议是：“不要通过共享内存来通信，而要通过通信来共享内存。”说白了就是：不要让几个 goroutine 读写同一个变量、再靠锁轮流访问，而是把数据沿着通道传下去，让同一时刻只有一个 goroutine 拥有它。发送方把值交出去，就不再碰它，接收方成为它的主人。别人都没有它，也就没什么可争的。这并不是禁止用锁。对计数器或缓存来说，互斥锁更简单，讲 sync 的那一部分会介绍。当数据从一个工作阶段流向下一个阶段，或者 goroutine 之间需要互相发信号时，再用通道。

要点

  • make(chan T) 是无缓冲的：发送要等接收方拿走值。make(chan T, n) 最多存放 n 个值，只有满了发送才等待。
  • 在函数签名里用 chan<- T 和 <-chan T。这样编译器会阻止接收方发送。
  • 如果所有 goroutine 都阻塞，Go 会以 all goroutines are asleep - deadlock! 停止。如果只是部分阻塞，什么警告都没有。
  • 只有发送方关闭通道。已关闭的通道永远返回零值和 ok == false，range 在关闭时停止，向已关闭的通道发送会 panic。
  • nil 通道永远阻塞。在 select 里，把通道设为 nil 就关掉了它的 case。
  • select 同时等待多个通道，在就绪的 case 中随机挑选。default 让它不阻塞，time.After 给它加上超时。
  • 关闭 done 通道，就能通知所有在听的 goroutine 停下。

在无缓冲通道上发送，要等到有人拿到值才算完成。

这篇文章对你有帮助吗？

点一颗爱心来评分！

平均评分 0 / 5. 投票总数： 0

还没有人投票。来做第一个评分的人吧。