八股Go 面试题
并发编程题
Go 面试中的并发控制、顺序打印、生产消费与超时实现。
如何用不超过 10 个 goroutine 处理 100 个元素？
回答重点
“最多 10 个 goroutine”和“最多 10 个任务同时执行”不是同一要求：
- 固定 worker pool 只创建 10 个工作 goroutine。
- 每个任务创建一个 goroutine 再用信号量限流，会创建 100 个 goroutine，只限制并发执行数。
题目明确限制 goroutine 数量时，应使用固定 worker pool：
package main
import (
"fmt"
"sync"
)
func main() {
const workerCount = 10
jobs := make(chan int)
var wg sync.WaitGroup
for workerID := 0; workerID < workerCount; workerID++ {
wg.Add(1)
go func(id int) {
defer wg.Done()
for value := range jobs {
fmt.Printf("worker=%d value=%d\n", id, value)
}
}(workerID)
}
for value := 0; value < 100; value++ {
jobs <- value
}
close(jobs)
wg.Wait()
}每个元素只会被一个 worker 接收，因此不会重复处理；输出顺序不保证与输入顺序一致。
如何让三个 goroutine 依次打印 abc，重复 10 次？
回答重点
为每个阶段使用一个 token channel，只有拿到 token 的 goroutine 才能输出，并把 token 传给下一阶段。
package main
import (
"fmt"
"sync"
)
func main() {
a := make(chan struct{})
b := make(chan struct{})
c := make(chan struct{})
var wg sync.WaitGroup
wg.Add(3)
go func() {
defer wg.Done()
for i := 0; i < 10; i++ {
<-a
fmt.Print("a")
b <- struct{}{}
}
}()
go func() {
defer wg.Done()
for i := 0; i < 10; i++ {
<-b
fmt.Print("b")
c <- struct{}{}
}
}()
go func() {
defer wg.Done()
for i := 0; i < 10; i++ {
<-c
fmt.Print("c")
if i < 9 {
a <- struct{}{}
}
}
}()
a <- struct{}{}
wg.Wait()
fmt.Println()
}最后一轮不能继续向 a 发送，否则没有接收者，会导致死锁。
如何用单个 Channel 让两个 goroutine 交替打印 0 和 1？
回答重点
可以让两个 goroutine 在同一个无缓冲 channel 上完成握手。打印 0 的 goroutine 先输出再发送，打印 1 的 goroutine先接收再输出；每轮末尾再反向确认。
package main
import (
"fmt"
"sync"
)
func main() {
turn := make(chan struct{})
var wg sync.WaitGroup
wg.Add(2)
go func() {
defer wg.Done()
for i := 0; i < 5; i++ {
fmt.Print(0)
turn <- struct{}{}
<-turn
}
}()
go func() {
defer wg.Done()
for i := 0; i < 5; i++ {
<-turn
fmt.Print(1)
turn <- struct{}{}
}
}()
wg.Wait()
fmt.Println()
}这个方案依赖两端操作顺序不对称。如果两个 goroutine 都先接收，它们会一起阻塞；如果由主 goroutine 先投递 token，则第一次由谁接收是不确定的。
如何用 sync.Cond 实现有界生产者消费者队列？
回答重点
sync.Cond 必须与锁和条件循环配合使用。Wait 会原子地解锁并挂起，返回前重新加锁；醒来后必须重新检查条件，因为条件可能已被其他 goroutine 改变。下面用同一个 Cond 等待“非空”和“未满”两个条件，因此状态变化后使用 Broadcast；生产环境也可以拆成两个条件变量，减少无效唤醒。
package main
import (
"fmt"
"sync"
)
type queue struct {
cond *sync.Cond
values []int
limit int
closed bool
}
func newQueue(limit int) *queue {
return &queue{
cond: sync.NewCond(&sync.Mutex{}),
limit: limit,
}
}
func (q *queue) put(value int) bool {
q.cond.L.Lock()
defer q.cond.L.Unlock()
for len(q.values) == q.limit && !q.closed {
q.cond.Wait()
}
if q.closed {
return false
}
q.values = append(q.values, value)
q.cond.Broadcast()
return true
}
func (q *queue) get() (int, bool) {
q.cond.L.Lock()
defer q.cond.L.Unlock()
for len(q.values) == 0 && !q.closed {
q.cond.Wait()
}
if len(q.values) == 0 {
return 0, false
}
value := q.values[0]
q.values = q.values[1:]
q.cond.Broadcast()
return value, true
}
func (q *queue) close() {
q.cond.L.Lock()
q.closed = true
q.cond.Broadcast()
q.cond.L.Unlock()
}
func main() {
q := newQueue(2)
go func() {
for i := 0; i < 5; i++ {
q.put(i)
}
q.close()
}()
for value, ok := q.get(); ok; value, ok = q.get() {
fmt.Println(value)
}
}如果只需要普通有界队列，带缓冲 channel 已经实现了生产者消费者同步，通常比 sync.Cond 更简单。
如何限制并发度并让整批任务最多执行 1 秒？
回答重点
使用固定数量的 worker 控制并发度，用 context.WithTimeout 通知生产者和 worker 停止。取消是协作式的，任务函数也必须检查 Context 才能及时退出。
package main
import (
"context"
"fmt"
"sync"
"time"
)
func main() {
ctx, cancel := context.WithTimeout(context.Background(), time.Second)
defer cancel()
jobs := make(chan int)
var wg sync.WaitGroup
for workerID := 0; workerID < 10; workerID++ {
wg.Add(1)
go func(id int) {
defer wg.Done()
for {
select {
case <-ctx.Done():
return
case job, ok := <-jobs:
if !ok {
return
}
fmt.Printf("worker=%d job=%d\n", id, job)
}
}
}(workerID)
}
sendLoop:
for job := 0; job < 1000; job++ {
select {
case <-ctx.Done():
break sendLoop
case jobs <- job:
}
}
close(jobs)
wg.Wait()
}原题常见错误是创建 1000 个 goroutine，却声称“控制 1000 个并发”。应先明确限制的是任务总数、goroutine 数量还是同时执行数。
如何让两个 goroutine 输出 a1b2c3？
回答重点
仍然用两个 token channel 表达轮次：
package main
import (
"fmt"
"sync"
)
func main() {
letterTurn := make(chan struct{})
numberTurn := make(chan struct{})
var wg sync.WaitGroup
wg.Add(2)
go func() {
defer wg.Done()
for letter := 'a'; letter <= 'c'; letter++ {
<-letterTurn
fmt.Printf("%c", letter)
numberTurn <- struct{}{}
}
}()
go func() {
defer wg.Done()
for number := 1; number <= 3; number++ {
<-numberTurn
fmt.Print(number)
if number < 3 {
letterTurn <- struct{}{}
}
}
}()
letterTurn <- struct{}{}
wg.Wait()
fmt.Println()
}面试官追问
- 如果某个 worker 提前返回，如何避免其余 goroutine 永久阻塞？
- 如何加入 Context 取消并保证所有 channel 只关闭一次？
- 输出顺序和任务完成顺序都要保持时，如何设计结果重排？
评论