← 文章 / 编程开发

Hacker News 2小时前 · 2026-09-27 19:29:39 · 1 阅读

Go 并发精粹

这本小书简要介绍了 Go 语言中众多并发主题。每个主题都配有交互式示例——你可以随意修改代码并点击 Run 进行实验。此外，还有包含静态示例的 PDF 版本。

这是 Go 并发知识的快速回顾，而非新手入门指南。如果你希望从零开始并通过实践练习来学习并发，可以参考我的另一本书—— Gist of Go: Concurrency。

本书内容不涉及 AI。

Goroutines • Channels • Select • Pipelines • Time • Context • Wait groups • Data races • Race conditions • Mutexes • Semaphores • Signaling • Run once • Object pool • Atomics • Testing • Scheduling • Diagnostics • Final thoughts

# Goroutines

Go 并发的基础是 goroutines（协程）——由 go 关键字启动的函数：

func main() {
    var wg sync.WaitGroup
    wg.Add(2)
    go func() {
        defer wg.Done()
        fmt.Println("worker 1")
    }()
    go func() {
        defer wg.Done()
        fmt.Println("worker 2")
    }()
    wg.Wait()
}

worker 2
worker 1

Go 运行时负责调度这些 goroutines，并将它们分配到运行在 CPU 核心上的操作系统线程中。与 OS 线程相比，goroutines 非常轻量，因此你可以轻松创建数百甚至数千个。

Goroutines 是完全独立的。主函数本身也是一个 goroutine，它在程序启动时隐式创建。当 main 结束时，其他 goroutines 也会随之终止。

上面例子中，我们用 wait group（sync.WaitGroup）等待 goroutine 执行完毕。它内部维护一个计数器：调用 Add(n) 会让计数器加 n，Done() 则减一。Wait() 会阻塞调用它的 goroutine（这里是 main），直到计数器归零。这样 main 就能等两个 worker 都完成后再退出。

WaitGroup.Go 会自动递增计数器、在 goroutine 里运行函数，并在结束时递减计数器：

func main() {
    var wg sync.WaitGroup
    wg.Go(func() {
        fmt.Println("worker 1")
    })
    wg.Go(func() {
        fmt.Println("worker 2")
    })
    wg.Wait()
}

worker 2
worker 1

# Channel

Goroutine 之间可以通过 channel 传递值。channel 就像一扇窗口：一个 goroutine 往里扔东西，另一个 goroutine 在外面接：

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

    go func() { messages <- "ping" }()

    msg := <-messages
    fmt.Println(msg)
}

ping

通过 channel 发送值是同步操作：发送方往 channel 写入值（ch <- val）后会阻塞，直到有人接收这个值（<-ch），然后才继续执行。

输出 channel

让函数返回一个输出 channel，并在内部 goroutine 中填充数据，是 Go 里常见的模式。这样调用方可以通过 channel 接收值，而函数自身保持对 channel 的控制权：

func generate(start, stop int) chan int {
    out := make(chan int)
    go func() {
        for i := start; i < stop; i++ {
            out <- i
        }
    }()
    return out
}

关闭 channel

要通知读取方数据已发送完毕，写入的 goroutine 可以用 close() 关闭 channel：

func generate(start, stop int) chan int {
    out := make(chan int)
    go func() {
        defer close(out)
        for i := start; i < stop; i++ {
            out <- i
        }
    }()
    return out
}

读取时，可以用第二个返回值（"comma OK"）检查 channel 的状态：

当通道处于打开状态时，接收方会获取下一个值以及 true 状态。如果通道已关闭，接收方将收到零值以及 false 状态。

通道只能关闭一次。再次关闭它或向已关闭的通道写入数据会导致 panic。

关闭通道的唯一理由是为了向读取者发出信号，表明所有数据已发送完毕。如果这对读取者不重要，则无需关闭通道。当通道不再使用时，Go 的垃圾回收器会释放其资源，无论通道是否关闭。

通道迭代

range 会自动读取通道中的下一个值并检查通道是否已关闭。如果通道已关闭，循环将退出：

func main() {
    nums := generate(5, 10)
    for n := range nums {
        fmt.Print(n, " ")
    }
}

5 6 7 8 9

对通道使用 range 仅返回单个值，而不同于对 slice 使用 range 返回的成对值。

单向通道

通过设置通道方向，可以防止误操作导致的写入或关闭错误。通道可以是：

chan（双向）：用于读取和写入（默认）；
chan<-（仅发送）：仅用于写入；
<-chan（仅接收）：仅用于读取。

不能从仅发送通道读取，也不能向仅接收通道写入（也不能关闭它）。

通道通常初始化为同时支持读写，而在函数参数中指定为单向通道。Go 会自动将普通通道转换为单向通道：

stream := make(chan int)

go func(in chan<- int) {
    in <- 42
}(stream)

func(out <-chan int) {
    fmt.Println(<-out)
}(stream)

带缓冲通道

带缓冲通道的工作方式类似于具有固定大小缓冲区用于存储值的 FIFO 队列。

只要缓冲区中有空闲空间，向通道写入数据就不会阻塞 goroutine。同样，只要缓冲区中包含值，从通道读取数据就不会阻塞 goroutine：

stream := make(chan int, 3)
stream <- 11
stream <- 12
stream <- 13

fmt.Println(<-stream)
fmt.Println(<-stream)

11
13

默认情况下，如果不指定缓冲区大小，通道就是无缓冲的（缓冲区大小为零）。

带缓冲的通道可以使用内置的 len() 和 cap() 函数：

stream := make(chan int, 3)
stream <- 11
fmt.Println(cap(stream), len(stream))

3 1

从已关闭的带缓冲通道中读取数据时，会先返回缓冲区内的值，并附带 true 状态。等缓冲区取空后，则返回零值和 false 状态，行为与普通通道一致：

stream := make(chan int, 1)
stream <- 11
close(stream)

val, ok := <-stream
fmt.Println(val, ok)
// 11 true

val, ok = <-stream
fmt.Println(val, ok)
// 0 false

11 true
0 false

nil 通道

和 Go 中的其他类型一样，通道的零值也是 nil。

向 nil 通道写入或从中读取，都会导致协程无限阻塞：

var stream chan int

go func() {
    // blocks forever
    stream <- 1
}()

// blocks forever
<-stream

关闭 nil 通道会引发 panic：

var stream chan int
close(stream)
// panic: close of nil channel

# Select

select 语句类似于 switch，但专门用于处理通道。其工作原理如下：

检查哪些分支未被阻塞。
若多个分支就绪，随机选择一个执行。
若所有分支都阻塞且存在默认分支，则执行默认分支。
若所有分支都阻塞且没有默认分支，则等待直到某个分支就绪。

Select 常用于管理流水线中的数据流：

// merge 将 in1 和 in2 的值发送到输出 channel。
func merge(in1, in2 <-chan int) <-chan int {
    out := make(chan int)
    go func() {
        defer close(out)
        for in1 != nil || in2 != nil {
            select {
            case val1, ok := <-in1:
                if ok { out <- val1 } else { in1 = nil }
            case val2, ok := <-in2:
                if ok { out <- val2 } else { in2 = nil }
            }
        }
    }()
    return out
}

// 假设我们向 in1 发送 10..12，向 in2 发送 20..22，
// 然后调用 merge(in1, in2)

10 11 20 12 21 22

取消 goroutine：

// process 处理 in 传来的值并发送到 out，
// 直到 in 耗尽或 cancel 被关闭。
func process(cancel chan struct{}, in <-chan int) <-chan int {
    out := make(chan int)
    go func() {
        for val := range in {
            select {
            case out <- val*10:
            case <-cancel:
                fmt.Println("canceled")
                return
            }
        }
    }()
    return out
}

// 假设我们向 in 发送 11 和 12，
// 然后调用 close(cancel)

110
120
canceled

非阻塞操作：

// multiplier 返回一个函数，该函数将输入乘以 10
// 并发送到 channel，
// 如果 channel 正忙则返回错误。
func multiplier(ch chan<- int) func(n int) error {
    return func(n int) error {
        select {
        case ch <- n*10:
            return nil
        default:
            return errors.New("busy")
        }
    }
}

func main() {
    nums := make(chan int, 1)
    multiply := multiplier(nums)

    err := multiply(11)
    fmt.Println(<-nums, err)
    // 110 <nil>

    err = multiply(12)
    fmt.Println(<-nums, err)
    // 120 <nil>

    err = multiply(13)
    err = multiply(14)
    fmt.Println(err)
    // busy
}

110 <nil>
120 <nil>
busy

以及更多场景。

# Pipelines（流水线）

流水线（pipeline） 是一系列操作组成的序列，每一步接收输入数据，按特定方式处理后输出。每个操作的输入和输出都是一个 channel。

典型的流水线是这样的：

读取器：从文件、数据库或网络读取输入数据。
N 个处理器：使用外部源对数据进行转换、过滤、聚合或增强。
写器：将处理后的数据写入文件、数据库或网络。

func read[T any]() <-chan T {
    out := make(chan T)
    go func() {
        defer close(out)
        for {
            // 从某处读取数据
            data := // ...
            out <- data
        }
    }()
    return out
}

func process[T any](in <-chan T) <-chan T {
    out := make(chan T)
    go func() {
        defer close(out)
        for inData := range in {
            // 处理数据
            outData = // ...
            out <- outData
        }
    }()
    return out
}

func write[T any](in <-chan T) <-chan struct{} {
    done := make(chan struct{})
    go func() {
        defer close(done)
        for data := range in {
            // 写入数据
        }
    }()
    return done
}

输出通道

一个 goroutine 可以通过输出通道向其他 goroutine 发送信号，表明其工作已完成：

func generate(start, stop int) <-chan int {
    out := make(chan int)
    go func() {
        defer close(out)
        for i := start; i < stop; i++ {
            out <- i
        }
    }()
    return out
}

func main() {
    nums := generate(5, 10)
    for n := range nums {
        fmt.Print(n, " ")
    }
}

5 6 7 8 9

完成通道

如果 goroutine 不需要返回结果，它可以利用完成通道来信号化其结束状态：

func work() <-chan struct{} {
    done := make(chan struct{})
    go func() {
        defer close(done)
        fmt.Println("work done")
    }()
    return done
}

func main() {
    done := work()
    <-done
}

work done

取消通道

若要提前终止一个 goroutine，调用方 goroutine 可以使用取消通道：

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

func main() {
    cancel := make(chan struct{})
    defer close(cancel)

    nums := generate(cancel, 10)
    fmt.Println(<-nums)
    fmt.Println(<-nums)
    fmt.Println(<-nums)
}

1
2
3

错误处理

在并发管道中，处理错误通常有三种思路。

➊ 遇到第一个错误就返回：

// calculate produces answers for the given numbers.
func process(in <-chan int) (<-chan int, <-chan error) {
	out := make(chan Answer)
	errc := make(chan error, 1)
	go func() {
		defer close(out)
		for n := range in {
			ans, err := fetchAnswer(n)
			if err != nil {
				errc <- err  // return with error
				return
			}
			out <- ans
		}
		errc <- nil          // return with nil
	}()
	return out, errc
}

➋ 使用结果类型：

// Result contains an answer or an error.
type Result struct {
	answer int
	err    error
}

// calculate produces answers for the given numbers.
func calculate(in <-chan int) <-chan Result {
	out := make(chan Result)
	go func() {
		defer close(out)
		for n := range in {
			ans, err := fetchAnswer(n)
			out <- Result{ans, err}  // return answer + error
		}
	}()
	return out
}

➌ 将错误收集在单独的通道中：

// calculate produces answers for the given numbers.
func calculate(in <-chan int, errc chan<- error) <-chan int {
	out := make(chan Answer)
	go func() {
		defer close(out)
		for n := range in {
			ans, err := fetchAnswer(n)
			if err == nil {
				out <- ans   // send answer
			} else {
				errc <- err  // or error
			}
		}
	}()
	return out
}

# 时间

除了处理日期和时间，time 包还为并发程序中的限时操作提供了工具。

After

time.After() 返回一个初始为空的通道，在超时时间结束后才会接收到值。它常用于操作超时控制：

// withTimeout 在指定超时时间内执行一个函数。
func withTimeout(timeout time.Duration, fn func()) error {
    done := make(chan struct{})
    go func() {
        defer close(done)
        fn()
    }()

    // 阻塞，直到 fn 完成或计时器到期，以先发生者为准
    select {
    case <-done:
        return nil
    case <-time.After(timeout):
        return errors.New("timeout")
    }
}

withTimeout() 会等待 fn() 执行完成，但借助 time.After()，等待时间不会超过 timeout：

func main() {
    var err error

    // 按时完成
    err = withTimeout(
        50*time.Millisecond,
        func() { fmt.Println("work done") },
    )
    fmt.Println("err =", err)

    // 超时被取消
    err = withTimeout(
        50*time.Millisecond,
        func() {
            time.Sleep(100 * time.Millisecond)
            fmt.Println("work done")
        },
    )
    fmt.Println("err =", err)
}

work done
err = <nil>
err = timeout

Timer

计时器（time.Timer）是一个带有 C 通道的结构体，触发（到期）时会向该通道发送当前时间。计时器常用于安排将来的执行：

done := make(chan struct{})

timer := time.NewTimer(50 * time.Millisecond)
go func() {
    eventTime := <-timer.C  // 阻塞 50ms
    fmt.Println("work done at", eventTime)
    close(done)
}()

<-done

work done at 2009-11-10 23:00:00.05

Stop() 用于停止计时器：如果它尚未到期则返回 true，否则返回 false：

// 计时器在 50ms 后到期
timer := time.NewTimer(50 * time.Millisecond)
go func() {
    eventTime := <-timer.C
    fmt.Println("work done at", eventTime)
}()

// 10ms 后计时器尚未到期
time.Sleep(10 * time.Millisecond)

if timer.Stop() {
    fmt.Println("execution canceled")
} else {
    fmt.Println("too late to cancel")
}

execution canceled

通常用 time.AfterFunc() 封装函数会更方便：它先等待时长 d，然后执行函数 f：

done := make(chan struct{})
work := func() {
    fmt.Println("work done")
    close(done)
}

// 50ms 后执行 work
time.AfterFunc(50*time.Millisecond, work)
<-done

work done

time.AfterFunc() 返回一个定时器，你可以在它开始执行之前取消它：

// 50ms 后执行函数
timer := time.AfterFunc(50*time.Millisecond, func() {})

// 10ms 后定时器尚未到期
time.Sleep(10 * time.Millisecond)

if timer.Stop() {
    fmt.Println("execution canceled")
}

execution canceled

如果在循环中使用定时器，最好只创建一个定时器并在每次迭代时重置它，避免每次创建新实例：

// 消费者从输入通道读取 token，
// 如果一个小时内通道中没有值，则发出警报。
func consumer(in <-chan token) {
    const timeout = time.Hour
    timer := time.NewTimer(timeout)
    for {
        timer.Reset(timeout)
        select {
        case <-in:
            // 处理数据
        case <-timer.C:
            // 记录警告
        }
    }
}

// 假设向 in 通道发送 10,000 个值
// 并测量内存占用。

Memory used: 4 KB, # allocations: 6

Ticker

Ticker（计时器）类似定时器，但会持续触发，直到你停止它。Ticker 适合执行周期性任务：

// 每 50ms 触发一次
ticker := time.NewTicker(50 * time.Millisecond)
defer ticker.Stop()

go func() {
    for {
        // 每次迭代等待 ticker 触发
        at := <-ticker.C
        fmt.Println("work done at", at)
    }
}()

// 给 ticker 足够的触发 3 次的时间
time.Sleep(160*time.Millisecond)
ticker.Stop()

work done at 2009-11-10 23:00:00.05
work done at 2009-11-10 23:00:00.10
work done at 2009-11-10 23:00:00.15

NewTicker(d) 创建 ticker，它会在间隔 d 内将当前时间发送到通道 C。最终必须用 Stop() 停止 ticker 以释放资源。

如果通道读取方跟不上 ticker 的速度，ticker 会跳过部分 tick。

# Context

context 的主要用途是取消操作，可以手动取消，也可以由超时/截止事件触发。

该函数接收一个 context，并通过其 `Done()` 通道监听取消信号：

// work 执行一项任务，持续 50 ms，除非被取消。
// 被取消时返回错误。
func work(ctx context.Context) error {
    done := make(chan struct{})

    go func() {
        time.Sleep(50 * time.Millisecond)
        fmt.Println("work done")
        close(done)
    }()

    select {
    case <-done:
        return nil
    case <-ctx.Done():
        return ctx.Err()
    }
}

手动取消（返回 `context.Canceled` 错误）：

func main() {
    // 空 context
    ctx := context.Background()
    // 手动取消 context
    ctx, cancel := context.WithCancel(ctx)
    defer cancel()

    done := make(chan struct{})
    go func() {
        // 持续 50 ms，除非被取消
        err := work(ctx)
        fmt.Println("err =", err)
        close(done)
    }()

    // 10 ms 后取消
    time.Sleep(10 * time.Millisecond)
    cancel()
    <-done
}

err = context canceled 按超时取消（返回 `context.DeadlineExceeded` 错误）：

func main() {
    ctx := context.Background()
    // 10 ms 后取消
    ctx, cancel := context.WithTimeout(ctx, 10*time.Millisecond)
    defer cancel()

    done := make(chan struct{})
    go func() {
        // 持续 50 ms，除非被取消
        err := work(ctx)
        fmt.Println("err =", err)
        close(done)
    }()

    <-done
}

err = context deadline exceeded 按截止时间取消（返回 `context.DeadlineExceeded` 错误）：

func main() {
    ctx := context.Background()
    // 在 现在 + 10 ms 时取消
    deadline := time.Now().Add(10 * time.Millisecond)
    ctx, cancel := context.WithDeadline(ctx, deadline)
    defer cancel()

    done := make(chan struct{})
    go func() {
        // 持续 50 ms，除非被取消
        err := work(ctx)
        fmt.Println("err =", err)
        close(done)
    }()

    <-done
}

err = context deadline exceeded Context 是分层级的，且 context 对象是不可变的。要为 context 添加新属性，需基于旧的（父）context 创建一个新的（子）context。父 context 和子 context 之间，较短的超时间隔总是优先生效。子 context 只能缩短父 context 的超时间隔，不能延长：

func main() {
    // 带有 100 ms 超时的父 context
    const dur100ms = 100 * time.Millisecond
    parentCtx, cancel := context.WithTimeout(context.Background(), dur100ms)
    defer cancel()

    // 带有 10 ms 超时的子 context
    const dur10ms = 10 * time.Millisecond
    childCtx, cancel := context.WithTimeout(parentCtx, dur10ms)
    defer cancel()

    // 现在工作会被取消
    err := work(childCtx)
    fmt.Println("err =", err)
}

err = context deadline exceeded

多次取消是安全的。你可以对 context 随意调用任意次 cancel()，第一次生效，后续调用会被忽略。

可以通过 context.WithCancelCause()、context.WithTimeoutCause() 和 context.WithDeadlineCause() 指定自定义的取消原因，再通过 context.Cause() 获取：

ctx, cancel := context.WithCancelCause(context.Background())
cancel(errors.New("the night is dark"))
fmt.Println(context.Cause(ctx))

the night is dark

还可以用 context.AfterFunc() 注册一个在 context 被取消时执行的函数：

ctx, cancel := context.WithCancel(context.Background())
cleanup := func() { fmt.Println("cleanup") }
context.AfterFunc(ctx, cleanup)
cancel()
time.Sleep(10 * time.Millisecond)

cleanup

Context 还可以用 context.WithValue() 传递额外信息，它会创建一个针对特定键携带值的 context。但通常应避免在 context 中传值，用显式参数或自定义结构体更好。

# Wait groups

sync.WaitGroup 类型可以等待一个或多个 goroutine 执行完毕：

const n = 10
var wg sync.WaitGroup
wg.Add(n)
for range n {
    go func() {
        defer wg.Done()
        fmt.Print(".")
    }()
}
wg.Wait()

..........

WaitGroup 对它所管理的 goroutine 一无所知，只靠一个内部计数器工作：调用 wg.Add(1) 将计数器加一，wg.Done() 将其减一，wg.Wait() 则会阻塞调用方 goroutine，直到计数器归零。

Go 方法将 Add、启动 goroutine 和 Done 结合起来：

var wg sync.WaitGroup
for range 10 {
    wg.Go(func() {
        fmt.Print(".")
    })
}
wg.Wait()

..........

所有方法都保证 goroutine 安全。

通常，所有 Add 调用都在 Wait 之前完成。但技术上，你完全可以让部分 Add 在 Wait 前调用，另一部分在 Wait 后（从另一个 goroutine）调用。

你可以从多个 goroutine 调用 Wait。它们都会阻塞，直到该组的计数器归零。

# 数据竞争

当多个 goroutine 访问共享数据，且至少其中一个修改了该数据时，就会发生数据竞争。我们需要保护数据免受这种并发访问的影响。

数据竞争不一定会导致运行时 panic。正因如此，Go 提供了一款名为 race detector 的专用工具。你可以使用 race 标志开启它，该标志与 test、run、build 和 install 命令配合使用。

var total int

// There's a data race on total.
var wg sync.WaitGroup
wg.Go(func() { total++ })
wg.Go(func() { total++ })
wg.Wait()

fmt.Println("total:", total)

total: 2

go run -race main.go

==================
WARNING: DATA RACE
...
2
Found 1 data race(s)

Channel 对并发读写是安全的，不会引发数据竞争。

防止数据竞争的方法：

避免并发修改数据（通常通过使用 channel）。
使用互斥锁来同步访问。
仅使用原子操作。

竞态条件

当来自多个 goroutine 的操作顺序不可预测，导致系统进入错误状态时，就会发生竞态条件：

// There's a race condition when working with balance.
withdraw := func(amount int) {
    if getBalance() < amount {
        return
    }
    time.Sleep(time.Millisecond)
    setBalance(getBalance() - amount)
}

setBalance(50)

var wg sync.WaitGroup
wg.Go(func() { withdraw(40) })
wg.Go(func() { withdraw(40) })
wg.Wait()

fmt.Println("balance:", getBalance())

balance: -30

Go 并发精粹

如果单项操作本身是并发安全的，Go 的竞态检测器（race detector）就不会发现问题。正因如此，它无法捕捉到竞态条件：

go run -race main.go

balance: -30

在并发环境中，不确定性无法完全消除。事件会以不可预测的顺序发生——这是并发的固有特性。不过，你可以通过保护复合操作来避免竞态条件，通常的做法是使用互斥锁（mutex）：

var mu sync.Mutex
withdraw := func(amount int) {
    mu.Lock()
    defer mu.Unlock()

    if getBalance() < amount {
        return
    }
    time.Sleep(time.Millisecond)
    setBalance(getBalance() - amount)
}

setBalance(50)

var wg sync.WaitGroup
wg.Go(func() { withdraw(40) })
wg.Go(func() { withdraw(40) })
wg.Wait()

fmt.Println("balance:", getBalance())

balance: 10

比较并交换

有时，你可以不依赖互斥锁，而是通过应用原子性的比较并交换（compare-and-set）操作或其变体来防止竞态条件：

// CompareAndSet changes the value to new if the current value equals old.
// Returns true if the value was changed.
CompareAndSet(old, new any) bool

// CompareAndSwap changes the value to new if the current value equals old.
// Returns the old value.
CompareAndSwap(old, new any) any

// CompareAndDelete deletes the value if the current value equals old.
// Returns true if the value was deleted.
CompareAndDelete(old any) bool

// etc

其核心思想始终一致：

检查假设的旧状态（old state）是否与现实相符。
如果相符，将状态更新为新值。
如果不符，不做任何操作。

# 互斥锁

sync.Mutex 类型用于保护共享数据及代码片段，防止它们被并发访问：

var total int
var mu sync.Mutex

var wg sync.WaitGroup
for range 100 {
    wg.Go(func() {
        mu.Lock()
        time.Sleep(time.Millisecond)
        total++
        mu.Unlock()
    })
}
wg.Wait()

total: 100

互斥锁保证在任意时刻，只有一个 goroutine 能执行 Lock() 与 Unlock() 之间的代码。

以下场景需要使用互斥锁：

当多个 goroutine 修改同一份数据时。
当一个 goroutine 修改数据，而其他 goroutine 正在读取该数据时。

如果所有 goroutine 都只读取数据，就不需要互斥锁。

TryLock

TryLock 方法会尝试锁定互斥锁，和普通的 Lock 一样。区别在于，如果锁不上，它会立即返回 false，而不是阻塞 goroutine：

var total int
var mu sync.Mutex

var wg sync.WaitGroup
for range 100 {
    wg.Go(func() {
        if !mu.TryLock() {
            return
        }
        defer mu.Unlock()
        time.Sleep(time.Millisecond)
        total++
    })
}
wg.Wait()

total: 1

RWMutex

sync.RWMutex 区分读操作和写操作，提供两组方法：

Lock / Unlock：以读写方式锁定和解锁互斥锁。
RLock / RUnlock：仅以只读方式锁定和解锁互斥锁。

var total int
var mu sync.RWMutex
var wg sync.WaitGroup

// 10 writers.
for range 10 {
    wg.Go(func() {
        mu.Lock()
        defer mu.Unlock()
        time.Sleep(time.Millisecond)
        total++
    })
}

// 10 readers.
for range 10 {
    wg.Go(func() {
        // Try switching from RLock/RUnlock to Lock/Unlock
        //and see how it affects the elapsed time.
        mu.RLock()
        defer mu.RUnlock()
        time.Sleep(time.Millisecond)
        _ = total
    })
}

wg.Wait()

elapsed: 10ms

它的规则如下：

如果某个 goroutine 用 Lock() 锁定了互斥锁，其他 goroutine 无论调用 Lock() 还是 RLock() 都会被阻塞。
如果某个 goroutine 用 RLock() 锁定了互斥锁，其他 goroutine 依然可以调用 RLock()，不会被阻塞。
只要有至少一个 goroutine 持有 RLock()，其他 goroutine 调用 Lock() 就会被阻塞。

这样就实现了“单写者、多读者”的模式。

Locker

sync.Mutex 和 sync.RWMutex 都实现了同一个 sync.Locker 接口：

type Locker interface {
    Lock()
    Unlock()
}

用 Locker 代替具体的互斥锁类型，就能编写不依赖具体锁实现的组件，把选哪种锁的决定权交给调用方。

Channel as mutex

你可以用 channel 代替互斥锁来保护共享数据：

var total int
lock := make(chan struct{}, 1)

var wg sync.WaitGroup
wg.Go(func() {
    lock <- struct{}{}
    defer func() { <-lock }()
    total++
})
wg.Go(func() {
    lock <- struct{}{}
    defer func() { <-lock }()
    total++
})
wg.Wait()

total: 2

# 信号量

信号量就像一个有 N 个可用槽位的容器，包含两种操作：acquire（获取）用于占据一个槽位，release（释放）用于腾出一个槽位。信号量的规则如下：

调用 acquire 会占据一个空闲槽位。
如果没有空闲槽位，调用 acquire 的 goroutine 会被阻塞。
调用 release 会腾出之前被占据的槽位。
如果 release 被调用时，有 goroutine 正阻塞在 acquire 上，其中一个会立即占据腾出的槽位并解除阻塞。

你可以用带缓冲的 channel 实现一个简单的信号量，其中 N 即为 channel 的容量。向 channel 发送值代表获取信号量，从 channel 接收值代表释放信号量：

// Try changing nConc and see how the elapsed time changes.
const nConc = 4
const nCalls = 100
sema := make(chan struct{}, nConc)

var wg sync.WaitGroup
for range nCalls {
    sema <- struct{}{} // acquire
    wg.Go(func() {
        defer func() { <-sema }() // release
        time.Sleep(time.Millisecond) // do some work
    })
}
wg.Wait()

elapsed: 25ms

对于更复杂的场景，可以使用 golang.org/x/sync/semaphore 包。

会合

会合（rendezvous）让两个 goroutine 互相等待：

有两个 goroutine——G1 和 G2，每个都可以发出自己已就绪的信号。
如果 G1 发出信号但 G2 尚未发出，G1 会被阻塞并等待。
如果 G2 发出信号但 G1 尚未发出，G2 会被阻塞并等待。
当两者都发出信号后，它们都会解除阻塞并继续执行。

你可以用 wait group 实现一个简单的会合：

var rend sync.WaitGroup
rend.Add(2)

var wg sync.WaitGroup
wg.Go(func() {
    fmt.Println("before rendezvous")
    rend.Done()
    rend.Wait()
    fmt.Println("after rendezvous")
})
wg.Go(func() {
    fmt.Println("before rendezvous")
    rend.Done()
    rend.Wait()
    fmt.Println("after rendezvous")
})
wg.Wait()

rendezvous 之前
rendezvous 之前
rendezvous 之后
rendezvous 之后

Barrier

Barrier 是 rendezvous 的通用形式。它允许 N 个 goroutine 相互等待：

Barrier 维护一个计数器（初始为 0）和一个阈值 N。
每个到达 Barrier 的 goroutine 将计数器加 1。
到达 Barrier 的 goroutine 会被阻塞。
当计数器达到 N 时，Barrier 解除所有等待 goroutine 的阻塞。

你可以用 WaitGroup 实现一个简单的 Barrier：

const n = 4
var bar sync.WaitGroup
bar.Add(n)

var wg sync.WaitGroup
for range n {
    wg.Go(func() {
        fmt.Println("barrier 之前")
        bar.Done()
        bar.Wait()
        fmt.Println("barrier 之后")
    })
}
wg.Wait()

barrier 之前
barrier 之前
barrier 之前
barrier 之前
barrier 之后
barrier 之后
barrier 之后
barrier 之后

# Signaling

sync.Cond（条件变量）类型允许一个 goroutine 向另一个发送“已就绪”信号，后者则等待该信号。

Cond 包含一个互斥锁，提供 Wait 和 Signal 两个方法。

Wait 解锁互斥锁并挂起 goroutine，直到收到信号。
Signal 唤醒在 Wait 中等待的 goroutine。
Wait 被唤醒后，会重新锁定互斥锁。

cond := sync.NewCond(&sync.Mutex{})
done := false

var wg sync.WaitGroup
wg.Go(func() {
    cond.L.Lock()
    fmt.Println("G1 已准备发送信号")
    done = true
    cond.Signal()
    cond.L.Unlock()
})
wg.Go(func() {
    cond.L.Lock()
    for !done {
        cond.Wait()
    }
    fmt.Println("G2 收到信号")
    cond.L.Unlock()
})
wg.Wait()

G1 已准备发送信号
G2 收到信号

若调用 Signal 时有多个 goroutine 在等待，只会恢复其中一个；若无等待的 goroutine，Signal 不做任何事。

也可以使用 Broadcast 方法。与 Signal 只唤醒一个在 Cond.Wait 上等待的 goroutine 不同，Broadcast 会唤醒所有此类 goroutine。

你也可以用 channel 来实现信号传递：

signal := make(chan struct{}, 1)

go func() {
    // do something
    signal <- struct{}{}
}()

go func() {
    <-signal
    // do something
}()

广播也可以：

broadcast := make(chan struct{})

go func() {
    // do something
    close(broadcast)
}()

go func() {
    <-broadcast
    // do something
}()

go func() {
    <-broadcast
    // do something
}()

条件变量的广播功能比较受限：它只发信号，不传实际数据，而且只能触发一次。用 channel 则可以搭出一个没有这些限制的发布/订阅系统：

type Publisher struct {
    sbox []chan int // subscription channels
    mu   sync.Mutex // protects the state
}

func (p *Publisher) Subscribe() <-chan int {
    p.mu.Lock()
    defer p.mu.Unlock()
    sub := make(chan int, 1)
    p.sbox = append(p.sbox, sub)
    return sub
}

func (p *Publisher) Broadcast(v int) {
    p.mu.Lock()
    defer p.mu.Unlock()
    for _, sub := range p.sbox {
        select {
        case sub <- v:
        default:
        }
    }
}

# 只执行一次

sync.Once 类型保证给定的函数只执行一次。如果多个 goroutine 同时调用 Once.Do，只有一个会真正执行该函数，其余的会等它返回后再继续：

total := 0
initState := func() {
    total += 1
}

var once sync.Once

var wg sync.WaitGroup
wg.Go(func() {
    once.Do(initState)
    // do something
})
wg.Go(func() {
    once.Do(initState)
    // do something
})
wg.Wait()

total: 1

在并发环境下做一次性初始化或清理，Once 非常合适。

除了 Once 类型，sync 包还提供了三个便捷的 once 函数：

// Calls f only once.
func (o *Once) Do(f func())

// Returns a function that calls f only once.
func OnceFunc(f func()) func()

// Returns a function that calls f only once
// and returns the value from that first call.
func OnceValue[T any](f func() T) func() T

// Returns a function that calls f only once
// and returns the pair of values from that first call.
func OnceValues[T1, T2 any](f func() (T1, T2)) func() (T1, T2)

# 对象池

sync.Pool 类型通过复用内存来替代每次新分配，从而减轻垃圾回收器的压力：

pool := sync.Pool{
    New: func() any {
        buf := make([]byte, 1024)
        return &buf
    },
}

// 尽管循环迭代了 4000 次，仅分配了 4*1024 字节。
var wg sync.WaitGroup
for range 4 {
    wg.Go(func() {
        for range 1000 {
            buf := pool.Get().(*[]byte)
            sink = buf
            pool.Put(buf)
        }
    })
}
wg.Wait()

内存分配量: 4 KB

Get 从池中取出一个项目。如果没有可用项目，则使用 New 创建新项（由于池子不了解其创建项目的具体情况，我们需要自行定义 New 逻辑）。Put 则将项目归还给池子。

需要留意以下几点：

New 应返回指针而非值，以减少内存拷贝并避免额外分配。
池没有大小限制。如果启动另外 1000 个 goroutine 且它们同时调用 Get，将额外分配 1000 个缓冲区。
一旦项目通过 Put 归还到池中，就不应再使用它（因为其他 goroutine 可能已经取走并开始使用）。

# 原子操作

没有同步机制的操作，只有当它能映射为单条处理器指令时，才是真正原子的。这类操作无需加锁，且并发调用时不会引发问题（即使是写操作）。

原子类型很少，且全部位于 sync/atomic 包中：

Int32     Bool
Int64     Value
Uint32    Pointer
Uint64

每种原子类型提供以下方法：

Load 读取变量的值。
Store 设置新值。
Swap 设置新值（类似 Store）并返回旧值。
CompareAndSwap 仅在当前值符合预期时才设置新值。

var n atomic.Int32
n.Store(10)
swapped := n.CompareAndSwap(10, 42)
fmt.Println("CompareAndSwap 10 -> 42:", swapped)
fmt.Println("n =", n.Load())

CompareAndSwap 10 -> 42: true
n = 42

数值类型还提供了 Add 方法，用于将值增加指定数额。

所有方法要么被翻译成单条 CPU 指令，要么保证原子性，因此可以安全地从多个 goroutine 中使用。

原子操作组合起来永远不是原子的：

var delta atomic.Int32
var counter atomic.Int32

func increment() {
    // 非原子操作，引发竞争条件。
    delta.Add(1)
    sleep(10)
    counter.Add(delta.Load())
}

// 在 100 次并发递增后，
// 最终值不保证正确。

counter = 9386

将组合操作变为原子操作并防止竞争条件最可靠的方法是使用互斥锁：

var delta int32
var counter int32
var mu sync.Mutex

func increment() {
    // 原子操作，不会引发竞争条件。
    mu.Lock()
    delta += 1
    sleep(10)
    counter += delta
    mu.Unlock()
}

// 在 100 次并发递增后，最终值得到保证：
// counter = 1+2+...+100 = 5050

counter = 5050

有时你可以使用原子类型替代互斥锁以实现提前退出：

type Gate struct {
    closed atomic.Bool
}

func (g *Gate) Close() {
    if !g.closed.CompareAndSwap(false, true) {
        return // 忽略重复调用
    }
    // 门已关闭。
    // 现在可以释放资源。
}

# 测试

如果你的并发程序使用了通道或带有同步方法（如 Wait）的自定义类型，你可以在测试中使用它们。这样，你的测试代码不会比同步代码复杂多少：

// Calc 异步计算某些内容。
func Calc() <-chan int {
    out := make(chan int, 1)
    go func() {
        out <- 42
    }()
    return out
}

func Test(t *testing.T) {
    // 等待 Calc goroutine 完成。
    got := <-Calc()
    if got != 42 {
        t.Errorf("got: %v; want: 42", got)
    }
}

PASS

如果你正在测试的代码中没有任何合适的同步“句柄”，可以使用 synctest 包。它导出了两个函数：

func Test(t *testing.T, f func(*testing.T))
func Wait()

synctest.Test 会运行一个隔离的 bubble。这个 bubble 使用虚拟时钟，你还可以通过 synctest.Wait 手动控制 goroutine 的同步。

synctest.Wait 会阻塞，直到 bubble 中所有 goroutine（除了调用 Wait 的那个）都已完成或处于持久阻塞状态。这样你就能等待某个 goroutine 阻塞或执行完毕，然后检查程序的状态：

// NewProc starts the calculation.
func NewProc() *Proc {
    p := &Proc{done: make(chan struct{})}
    go func() {
        p.res = 42
        <-p.done // (X)
        p.res = 0
    }()
    return p
}

func Test(t *testing.T) {
    synctest.Test(t, func(t *testing.T) {
        p := NewProc()
        defer p.Stop()

        // Wait for the goroutine to block at point X.
        synctest.Wait()
        if got := p.Res(); got != 42 {
            t.Fatalf("got %v, want 42", got)
        }
    })
}

PASS

synctest.Test 的虚拟时钟只有同时满足以下条件才会推进：➊ bubble 中所有 goroutine 都处于持久阻塞状态；➋ 未来某个时刻至少有一个 goroutine 会被解除阻塞；➌ synctest.Wait 没有在运行。有了这个机制，涉及时间的测试可以瞬间跑完：

// Calc processes a value from the input channel.
// Times out if no input is received after 3 seconds.
func Calc(in chan int) (int, error) {
    select {
    case v := <-in:
        return v * 2, nil
    case <-time.After(3 * time.Second):
        return 0, ErrTimeout
    }
}

func Test(t *testing.T) {
    synctest.Test(t, func(t *testing.T) {
        ch := make(chan int)
        got, err := Calc(ch) // runs instantly

        if err != ErrTimeout {
            t.Errorf("got: %v; want: %v", err, ErrTimeout)
        }
        if got != 0 {
            t.Errorf("got: %v; want: 0", got)
        }
    })
}

PASS

以下操作会让 goroutine 进入持久阻塞状态：

在 bubble 内创建的 channel 上进行阻塞式的发送或接收。
select 语句的所有 case 都操作 bubble 内创建的 channel 且发生阻塞。
调用 Cond.Wait。
调用 WaitGroup.Wait，前提是所有 WaitGroup.Add 调用都发生在 bubble 内。
调用 time.Sleep。

阻塞在互斥锁、I/O 或系统调用上不被视为持久状态，synctest 测试环境也无法处理这些情况。

# 调度

在硬件层面，CPU 核心负责执行并行任务。

在操作系统层面，线程是执行的基本单元。通常线程数量远多于 CPU 核心数，因此操作系统调度器会决定运行哪些线程、暂停哪些线程。

在 Go 运行时层面，Goroutine 是执行的基本单元。运行时调度器运行固定数量的操作系统线程，通常每个 CPU 核心对应一个线程。由于 Goroutine 的数量可以远超线程数，调度器会决定哪些 Goroutine 在可用线程上运行，哪些需要暂停。调度器通过不断切换 Goroutine，确保每个 Goroutine 都能轮到在某个线程上运行，而不是永远排队等待。

  CPU                  OS                   Go runtime
┌──────────┐  run on ┌──────────┐  run on ┌────────────┐
│ Cores    │ <────── │ Threads  │ <────── │ Goroutines │
└──────────┘         └──────────┘         └────────────┘

这就是 Go 处理并发的方式。

Goroutine 调度器

Goroutine 调度器的任务是在 N 个操作系统线程上运行 M 个 Goroutine，其中 M 可以远大于 N。以下是其算法的简化版本：

如果有空闲线程，从队列中分配一个 Goroutine 给它。
如果正在运行的 Goroutine 发生阻塞（例如从 channel 读取数据时），将其放回队列，并为该线程分配另一个 Goroutine。
如果正在运行的 Goroutine 陷入系统调用，启动一个新线程来运行其他 Goroutine，直到阻塞的 Goroutine 完成系统调用。
每 10 ms 检查一次正在运行的 Goroutine。抢占长时间运行的 Goroutine 并将其返回队列，以防止饿死现象。

┌─────┐┌─────┐┌─────┐┌─────┐
│ G17 ││ G18 ││ G19 ││ G20 │                        队列
└─────┘└─────┘└─────┘└─────┘

┌─────┐      ┌─────┐      ┌─────┐      ┌─────┐
│ G15 │      │ G16 │      │ G13 │      │ G14 │      运行中
└─────┘      └─────┘      └─────┘      └─────┘
  │            │            │            │
┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐
│ Thread E │ │ Thread F │ │ Thread C │ │ Thread D │
└──────────┘ └──────────┘ └──────────┘ └──────────┘

┌─────┐      ┌─────┐
│ G11 │      │ G12 │                                系统调用
└─────┘      └─────┘
  │            │
┌──────────┐ ┌──────────┐
│ Thread A │ │ Thread B │
└──────────┘ └──────────┘

运行 Go 代码的线程数由环境变量 GOMAXPROCS 或函数 runtime.GOMAXPROCS 控制。

每个 Goroutine 的初始内存占用约为 2 KB，主要用于栈空间。栈大小可以根据需要动态增长。由于 Goroutine 极其轻量，即使在小机器上也能轻松运行数万甚至数十万个。

# 诊断

在生产环境中排查并发程序问题时，我们通常借助指标（Metrics）、性能分析（Profiling）和追踪（Tracing）。

指标展示了 Go 运行时的性能状况，例如堆内存用量或垃圾回收暂停时长。每项指标都有唯一的名称和数值，数值可以是数字，也可以是直方图。

你可以使用 runtime/metrics 包获取完整的指标列表，或查询特定指标的值：

samples := []metrics.Sample{
    {Name: "/sched/gomaxprocs:threads"},
    {Name: "/sched/goroutines:goroutines"},
}
metrics.Read(samples)

for _, s := range samples {
    fmt.Printf("%s: %v\n", s.Name, s.Value.Uint64())
}

/sched/gomaxprocs:threads: 8
/sched/goroutines:goroutines: 1

在实际应用中，很少有人会手动执行此操作。通常的做法是通过 Prometheus 或 OpenTelemetry 库自动导出所有指标。

性能分析有助于理解程序具体在执行什么操作、消耗了哪些资源，以及这些行为发生在代码的哪些位置。Go 使用的是适合生产环境的采样式性能分析工具。

最常用的 profile 有两种：CPU profile 显示每个函数占用了多少处理器时间，heap profile 显示每个函数占用了多少堆内存。此外，goroutine、block 和 mutex profile 可以帮助排查并发相关的问题。

给应用添加性能分析最简单的方式是使用 net/http/pprof 包。要采集指定名称的 profile，访问 /debug/pprof/{name} 端点即可。要查看采集到的 profile，可以使用 go tool pprof 工具：

go tool pprof -proto \
  "http://localhost:6060/debug/pprof/profile?seconds=N" > cpu.pprof
go tool pprof -http=localhost:8080 cpu.pprof

也可以手动采集 profile：

// CPU profile。
file, _ := os.Create("cpu.prof")
defer file.Close()
pprof.StartCPUProfile(file)
defer pprof.StopCPUProfile()
// ...

// 其他任意 profile。
file, _ := os.Create(name + ".prof")
defer file.Close()
pprof.Lookup(name).WriteTo(file, 0)

Tracing（追踪） 会在程序运行时记录特定类型的事件，主要涉及并发和内存。当 net/http/pprof 包的分析服务器运行时，访问 /debug/pprof/trace 端点即可采集 trace。查看结果可以使用 go tool trace 工具。

也可以手动采集 trace：

file, _ := os.Create("trace.out")
defer file.Close()
trace.Start(file)
defer trace.Stop()
// ...

你还可以设置一个按大小或时长限制的滑动窗口来自动追踪，这被称为"flight recording"（飞行记录）。这样就能始终保留最近一段 trace，出了问题随时可查：

cfg := trace.FlightRecorderConfig{
    MinAge:   5 * time.Second,
    MaxBytes: 3 << 20, // 3MB
}
rec := trace.NewFlightRecorder(cfg)
rec.Start()
defer rec.Stop()

# 结语

本文介绍了许多用于编写并发程序的 Go 工具：

Goroutine：运行并发任务。
Channel 和 select：灵活的通信工具。
Timer 和 ticker：处理时间相关操作。
Context：取消操作。
Wait group：同步 goroutine。
Mutex：防止竞态条件。
条件变量（condition variable）：事件通知。
Once：安全的一次性初始化。
使用对象池降低垃圾回收器的负担。
原子操作。

如果你喜欢这本书，欢迎推荐给身边的朋友或同事。感兴趣的话，也可以看看我的其他书籍和项目。

很高兴你能读完这本书。谢谢，下次再见！

原始来源： Hacker News