请求 A
尚未返回- 读取n = stock.Load()
- 检查n > 0？
- 扣减Add(−1)
- 返回true / false
本请求保存的局部 n尚未读取共享库存变了，这个局部值不会自动跟着变。
- 最近一次检查
- 待检查
- 已尝试扣减
- 0 次
当前 / 下一步Load 读取库存
主题
sync 包：从保护一组状态，到等一批任务结束
sync 里有锁，也有计数器、一次性初始化和临时对象池。它们解决的问题不同，不能因为都带“并发安全”就相互替代。最省力的学法，是先说清自己要守住什么约束，再选对应的工具。
本文使用 Go 1.27.2。OnceFunc、OnceValue、OnceValues 从 Go 1.21 起可用；WaitGroup.Go 从 Go 1.25 起可用。文末模块声明 Go 1.25，因此示例可以使用后者。类型化原子值在 sync/atomic，不是 sync 的子类型。[1][2]
先按工作选择入口
| 想做的事 | 常用入口 | 它不负责的事 |
|---|---|---|
| 同时修改几个有关联的字段 | Mutex.Lock/Unlock | 不自动选择锁的范围，也不替你防死锁 |
| 让多个只读操作并行 | RWMutex.RLock/RUnlock，写入用 Lock/Unlock | 不保证比 Mutex 更快，读锁里也不能写 |
| 等一组任务结束 | WaitGroup.Go/Wait，旧式 Add/Done/Wait | 不取消任务，不收集错误 |
| 初始化一次并安全发布结果 | Once.Do、OnceValue 等 | 不重试失败，不等于每次调用都执行 |
| 等共享状态满足条件 | Cond.Wait/Signal/Broadcast | 不保存业务条件，不自动支持 context |
| 重用短期对象，减少分配 | Pool.Get/Put，可选 New | 不保证对象一定还在，不适合管理必须关闭的资源 |
| 某些模式下的并发键值表 | Map.Load/Store/LoadOrStore 等 | 不保护值内部对象，也不保证多个键原子更新 |
| 单个计数、标志或指针的原子操作 | atomic.Int64、Bool、Pointer[T]、Value | 多步业务操作不会自动组成一个事务 |
这些类型大多零值可用，但 Cond 要关联一把 Locker，Pool 要决定没有现成对象时怎么办。sync 中的同步值不应在首次使用后复制；原子类型也有类似约束。结构体包含它们时，同样要避免值接收者、按值传参、赋值、append 等造成的意外复制。[1:1][3][2:1]
Mutex：锁住的是约定，不是字段本身
计数器最直接的写法，是读写都经过同一个锁：
go
func (c *Counter) Add() {
c.mu.Lock()
defer c.mu.Unlock()
c.n++
}
func (c *Counter) Value() int {
c.mu.Lock()
defer c.mu.Unlock()
return c.n
}启动 100 个任务调用 Add，等它们都完成，再读取，本次输出：
text
counter: 100如果只给写操作加锁，却让读取直接访问 c.n，读写仍可能产生数据竞争。锁不会替你查找所有共享字段，必须由代码共同遵守“访问这份状态先拿这把锁”的约定。Unlock 与后续成功 Lock 之间建立同步关系，使前一个临界区的写入可被后一个看见。[4][5]
当状态包括 balance 和 reserved 时，重点不只是两次赋值各自安全，而是外部不能看见只更新一半的状态。因此应把一次业务变更的检查和修改放在同一个临界区。若把检查和扣减分成两个各自上锁的方法，两个请求仍可能都通过检查。这是业务竞态，race 检测未必会报。
锁的几个边界
- Mutex 不可重入。同一个 goroutine 已持锁时再次
Lock，若没有其他路径解锁，会把自己卡住；它不记录“当前调用者已经有锁”。 Unlock未加锁的 Mutex 是运行时错误。锁不绑定到特定 goroutine，因此一个 goroutine 加锁、另一个解锁在 API 上允许，但生命周期会更难审查。TryLock只尝试一次；失败没有建立任何同步关系，不能据此安全读取锁保护的数据。- 不要把潜在长等的 I/O、用户回调和复杂下游调用随意放进临界区。否则其他访问者要跟着等待，还可能形成锁顺序环。这是缩小等待依赖的设计建议。[4:1]
RWMutex：读多只是候选条件，不是性能结论
| 行为 | Mutex | RWMutex |
|---|---|---|
| 多个读者同时进入 | 不允许 | 可同时持有读锁 |
| 有写者持锁时 | 其他访问等待 | 其他读写都等待 |
| 写者已等待时的新读者 | 没有单独的读写分类 | 会被阻止继续拿读锁，以便写者推进 |
| 适合先尝试的场景 | 临界区短，读写混合，逻辑简单 | 读操作确实占大多数，且并行读足以抵消额外协调成本 |
RLock 不能升级成写锁；写锁也不能直接降级成读锁。递归拿读锁同样危险：如果中间有写者开始等待，第二次 RLock 可能阻塞，而写者又等第一次读锁释放。读锁不是“可以随便再拿一次”的轻量标记。[6]
是否更快要测自己的临界区长度、读写比例和并发程度。本文没有做 Mutex 与 RWMutex 的性能基准，不给固定阈值。
WaitGroup：把“都结束了”做成一个可靠边界
Go 1.25 以后，最短的正常用法是：
go
var wg sync.WaitGroup
for range n {
wg.Go(c.Add)
}
wg.Wait()Go 同时登记任务和启动 goroutine，函数返回后任务从计数中移除。文档要求传入的函数不能 panic；不要把 WaitGroup 当成异常恢复器。这个例子就是前面 counter: 100 的启动部分。[7]
旧式写法仍会遇到：先 Add(1)，再 go，goroutine 内部 defer Done()。不要把 Add 放进新 goroutine：主线程可能在 Add 发生前就调用 Wait，看见零任务而直接返回。计数降成负数会 panic；重新使用同一个 WaitGroup 开始新一轮任务，必须在上一轮 Wait 都已返回之后。[7:1]
Wait 返回是一个同步边界，能用来读取已经完成任务的结果。但“多个 goroutine 往同一个 slice append，再等它们结束”仍然不安全：Wait 只连接任务与等待者，不会把任务之间相互重叠的写变成串行。可以给每个任务分配独立下标，或者把结果发送到单一收集者。
Once：执行一次，连失败语义也只认一次
| 入口 | 得到什么 | 函数 panic 后再次调用 |
|---|---|---|
once.Do(f) | 保证只执行一回 f | 不再执行 f；这次调用已计作完成 |
OnceFunc(f) | 一个可重复调用的包装函数 | 每次都再次 panic，值与首次相同 |
OnceValue(f) | 每次返回同一个单值结果的函数 | 每次都再次 panic |
OnceValues(f) | 每次返回同一对结果的函数 | 每次都再次 panic |
Once 会让并发调用者等初始化完成，不是用一个普通 bool 提前标记“开始初始化”。f 返回与相关 Do 返回之间有同步保证，适合安全发布初始化结果。[8]
go
initCount := 0
config := sync.OnceValue(func() string {
initCount++
return "ready"
})
fmt.Println(config(), config(), "initializations:", initCount)实际输出：
text
ready ready initializations: 1若用 OnceValues(func() (Config, error)) 读配置，第一次返回的 error 也会被记住。需要临时错误后重试时，应另写重试和状态机，不能期望 Once 自动重新初始化。Do 的回调再调用同一个 Once.Do 还会等待自己，形成死锁。本文测试另外验证了 Once.Do panic 后不重跑、OnceValue 重复抛出同一 panic 的差异。[8:1]
Cond：等待的主角是条件，通知只是催你再看一眼
channel 能表达“收到一个值”或“流已关闭”。当多个操作需要等待共享状态的某种组合时，例如“队列非空，或者队列已关闭”，可以考虑 Cond。条件由共享字段表达，Cond 负责等和唤醒。[3:1]
本文 Mailbox 持有 queue 和 closed，它们都受 mu 保护。接收操作是：
go
m.mu.Lock()
defer m.mu.Unlock()
for len(m.queue) == 0 && !m.closed {
m.ready.Wait()
}
if len(m.queue) == 0 {
return 0, false
}Wait 调用前必须持有 Cond 的锁；它原子地解锁并暂停，醒来后重新加锁，才返回给调用者。这样等待者不会一直占着锁，让生产者永远无法改变条件。[3:2]
正在绘制图示…
查看图源代码
sequenceDiagram
participant C as 接收者
participant L as 同一把锁与共享状态
participant P as 发送者
C->>L: Lock，发现队列空且未关闭
C->>L: Wait：释放锁，进入等待
P->>L: Lock，追加元素
P->>C: Signal：唤醒一个等待者
P->>L: Unlock
C->>L: Wait 返回前重新拿锁
C->>L: 再检查条件，取出元素，Unlock
为什么用 for 而不是 if？不是因为 Go 的 Cond 会无故返回。文档说明 Wait 只在 Signal/Broadcast 后返回；真正的问题是被唤醒到重新拿到锁之间，其他 goroutine 可能已经改变了条件。通知不把某个元素预留给被唤醒者。[3:3]
发送后 Signal() 唤醒一个等待者；关闭邮箱后 Broadcast() 唤醒所有等待者，让它们重新检查结束条件。本文发送 7，再关闭邮箱，先读到数据，第二次接收结束：
text
mailbox: 7 true false示例邮箱是无容量限制、单消费者的教学实现，没有模拟完整生产队列。若队列有容量限制，需要再定义“可写”条件和发送者等待策略；若需要取消，还得安排取消时唤醒和在锁内检查的协议。Cond 没有自带带 context 的 Wait。简单一次性通知，关闭 channel 往往更直接。
Pool：借出临时对象，归还后停止使用
go
pool := sync.Pool{New: func() any { return new(bytes.Buffer) }}
buf := pool.Get().(*bytes.Buffer)
buf.Reset()
buf.WriteString("temporary")
fmt.Println(buf.String())
buf.Reset()
pool.Put(buf)实际输出：
text
temporaryGet 可以返回池中的任意对象，也可能无视池中已有对象；没有对象且设置了 New 才调用它。池中对象可随时被自动移除，不能靠 Pool 保证缓存命中或固定数量。它主要用来复用临时对象、减少分配压力。[9]
在 Put(buf) 后继续写 buf，或者继续使用 buf.Bytes() 得到的共享切片，都可能与下一个借用者冲突。归还时的关键是所有权已经交走。重置能避免下一位读到上一次内容，但 Buffer.Reset 不会必然归还它的大容量底层内存；是否丢弃过大的 buffer，需要根据负载设策略。[9:1][10]
数据库连接和文件需要容量限制、健康检查、显式关闭。Pool 不承诺归还通知或淘汰回调，因此不能拿它直接替代这类资源池。本文只验证借还使用方式，没有测分配率或吞吐提升。
Map：一条原子操作能解决的事，别拆成两条
普通 map 加锁通常更容易维护类型和多个字段之间的约束。sync.Map 针对特定模式优化：一个键写一次、读多次；或者不同 goroutine 操作大体互不相交的键集合。是否适合其他负载要测试。[11]
| 动作 | 入口 | 需要留意 |
|---|---|---|
| 读、写、删 | Load、Store、Delete | 返回值是 any，需要类型约定 |
| 不存在才放入 | LoadOrStore | 返回实际值和“原来已存在”的标志 |
| 读取并删除 | LoadAndDelete | 一步完成，不用先 Load 再 Delete |
| 交换值 | Swap | 返回旧值及是否存在 |
| 按旧值比较后更新/删除 | CompareAndSwap、CompareAndDelete | 用于比较的旧值必须可比较 |
| 遍历、清空 | Range、Clear | Range 不是一致性快照；Clear 不把别处持有的对象销毁 |
go
var cache sync.Map
a, loadedA := cache.LoadOrStore("key", 7)
b, loadedB := cache.LoadOrStore("key", 9)
fmt.Println("map:", a, loadedA, b, loadedB)实际输出：
text
map: 7 false 7 true第一次放入 7，第二次拿到已有 7，9 没替换它。若写成 Load 发现不存在，再 Store，两步之间另一个 goroutine 可能已写入，语义就不同了。[11:1]
Map 只保护它自己的键值操作。值是 *User 时，两个 goroutine 从 Map 取出同一个指针后同时改 User.Name，仍需额外同步。Range 期间并发写入时，同一轮可能看见不同时间点的键值状态，不能据此计算需要严格一致的财务总额或业务快照。[11:2]
atomic：单个操作可靠，组合逻辑还要自己证明
新代码通常先看类型化入口：atomic.Int64、Uint64 等有 Load、Store、Add、Swap、CompareAndSwap；Bool 处理标志，Pointer[T] 处理指针，Value 保存类型一致的任意值。零值可以作为相应原子对象的初始状态，但 Value 首次 Store 之前 Load 返回 nil，Store nil 或改变已存储的具体类型会 panic。[2:2]
go
var completed atomic.Int64
completed.Add(1)
fmt.Println("atomic:", completed.Load())实际输出：
text
atomic: 1原子操作按 Go 内存模型表现为一个顺序一致的顺序，但这不把 Load 加 Add 自动合并。例如库存是 1，两个请求都 Load 得到 1，再各 Add(-1)，就会超卖。要么把检查和扣减放在一把锁内，要么通过 CAS 循环把“仍等于我看到的值才扣减”做成一个不可拆开的更新；多字段约束通常用锁更容易看懂。[2:3][5:1]
发布指针时也是如此：atomic.Pointer 可以安全更新“当前配置”的指针，但已经发布的配置若继续被随意修改，仍会出问题。常见设计是发布不可变快照。无锁只是没有显式互斥锁，不等于没有竞争成本，也不保证算法更快或没有饥饿。
交互实验：原子操作与库存约束
每次读写都是原子的，为什么还会超卖？
请求 A、B 各想预留一件商品。你决定谁先执行下一步，观察“检查有库存”和“真正扣减”之间的空隙，再用 CAS 循环比较同一次交错。
01
输入 · 选实现，再设初始库存
要守住的约束成功预留数 ≤ 初始库存等价地，共享库存始终 ≥ 0。单次 Add 原子性不等于完整预留操作的原子性。
保留当前实现与有效库存，清空事件和请求状态。
从一个明确的交错开始
02
过程 · 选择下一个请求
第 0 步 · 初始库存为 1。两个请求都还没读取；选一个请求推进。
当前共享库存 stock1只有 Add 或成功的 CAS 会改变它
请求 B
尚未返回- 读取n = stock.Load()
- 检查n > 0？
- 扣减Add(−1)
- 返回true / false
本请求保存的局部 n尚未读取共享库存变了，这个局部值不会自动跟着变。
- 最近一次检查
- 待检查
- 已尝试扣减
- 0 次
当前 / 下一步Load 读取库存
每次只推进所选请求的一步。Load 和 Add / CAS 各自是原子操作；中间的检查只读本请求的局部 n。按 Tab 选按钮，Enter 或空格执行。
事件日志 · 0 步，按实际点击顺序
尚未执行。预设自动完成的步骤也会完整显示在这里。
03
结果 · 没有 data race，也可能做错生意
进行中：尚未超卖还有 2 个请求没结束
成功预留0 / 2预留操作最终返回 true 的请求数
拒绝预留0 / 2重新检查到无库存才返回 false；CAS 失败本身不算拒绝
超卖件数0成功预留数超出初始库存的部分
试试“交错：都通过检查”，再分别推进 A、B 扣减。 对比“串行”预设：一次安全结果不能证明所有交错都安全。
模型边界与对应的 Go 写法
这是确定性的教学交错，浏览器没有运行 Go。按钮指定了一个合法的操作顺序；预设也按写明的顺序执行。它不模拟运行时调度概率、CPU 耗时或吞吐量。
拆开模式对应先读取 n := stock.Load()，检查 n > 0，再执行 stock.Add(-1)。CAS 模式对应循环读取、检查，并调用 stock.CompareAndSwap(n, n-1)；只有 CAS 成功才返回 true，CAS 失败回到循环开头。
这里的共享库存访问全部对应 atomic.Int64 操作，局部 n 属于各自请求。data race 与业务竞态是不同问题；race 检测不负责报告这种超卖。涉及库存、订单和预留量等多个字段时，通常应考虑用同一把锁保护完整约束。这个实验不证明任意无锁算法都正确或更快。
不操作也能读懂：库存 1，A、B 都先读到 1；拆开扣减会得到 −1。换成 CAS，一个成功后，另一个 CAS 失败，重读到 0，再检查并拒绝，最终库存为 0。
依据：Go 内存模型：原子操作；sync/atomic：CompareAndSwap（核对日期：2026-10-14）。
依据与本次验证
Go 1.27.2、Linux/amd64 上，程序实际输出上面的 counter、OnceValue、atomic、buffer、Map 和 Mailbox 结果。五项测试覆盖计数、并发 OnceValue、邮箱排空及两种 panic 语义；go test -race -count=1 ./...、go vet ./... 通过。RWMutex 性能、Pool 的分配收益和 sync.Map 的吞吐差异均未测量，本文没有把适用条件写成性能保证。
完整代码与复现方法
以同一目录为根，按下面标出的相对路径保存文件。使用 Go 1.27.2 运行：
sh
go run .
go test -race -count=1 ./...
go vet ./...go.mod
go.mod
module example.com/go-articles/04_sync
go 1.25.0main.go
代码块 · 95 行 · 点击收起 / 展开
go
package main
import (
"bytes"
"fmt"
"sync"
"sync/atomic"
)
type Counter struct {
mu sync.Mutex
n int
}
func (c *Counter) Add() { c.mu.Lock(); defer c.mu.Unlock(); c.n++ }
func (c *Counter) Value() int { c.mu.Lock(); defer c.mu.Unlock(); return c.n }
// Mailbox has one consumer and is closed after its last Send.
type Mailbox struct {
mu sync.Mutex
ready *sync.Cond
queue []int
closed bool
}
func NewMailbox() *Mailbox {
m := &Mailbox{}
m.ready = sync.NewCond(&m.mu)
return m
}
func (m *Mailbox) Send(n int) {
m.mu.Lock()
defer m.mu.Unlock()
if m.closed {
panic("send after mailbox close")
}
m.queue = append(m.queue, n)
m.ready.Signal()
}
func (m *Mailbox) Close() {
m.mu.Lock()
m.closed = true
m.ready.Broadcast()
m.mu.Unlock()
}
func (m *Mailbox) Receive() (int, bool) {
m.mu.Lock()
defer m.mu.Unlock()
for len(m.queue) == 0 && !m.closed {
m.ready.Wait()
}
if len(m.queue) == 0 {
return 0, false
}
n := m.queue[0]
m.queue = m.queue[1:]
return n, true
}
func Count(n int) int {
var c Counter
var wg sync.WaitGroup
for range n {
wg.Go(c.Add)
} // Go 1.25+; callback must not panic.
wg.Wait()
return c.Value()
}
func main() {
fmt.Println("counter:", Count(100))
initCount := 0
config := sync.OnceValue(func() string { initCount++; return "ready" })
fmt.Println(config(), config(), "initializations:", initCount)
var completed atomic.Int64
completed.Add(1)
fmt.Println("atomic:", completed.Load())
pool := sync.Pool{New: func() any { return new(bytes.Buffer) }}
buf := pool.Get().(*bytes.Buffer)
buf.Reset()
buf.WriteString("temporary")
fmt.Println(buf.String())
buf.Reset()
pool.Put(buf) // No use of buf after ownership is returned.
var cache sync.Map
a, loadedA := cache.LoadOrStore("key", 7)
b, loadedB := cache.LoadOrStore("key", 9)
fmt.Println("map:", a, loadedA, b, loadedB)
m := NewMailbox()
var wg sync.WaitGroup
wg.Go(func() { m.Send(7); m.Close() })
n, ok := m.Receive()
_, more := m.Receive()
wg.Wait()
fmt.Println("mailbox:", n, ok, more)
}main_test.go
代码块 · 78 行 · 点击收起 / 展开
go
package main
import (
"sync"
"sync/atomic"
"testing"
)
func TestCount(t *testing.T) {
if got := Count(1000); got != 1000 {
t.Fatal(got)
}
}
func TestOnceValue(t *testing.T) {
var calls atomic.Int64
f := sync.OnceValue(func() int { calls.Add(1); return 7 })
var wg sync.WaitGroup
for range 20 {
wg.Go(func() {
if f() != 7 {
t.Error("bad value")
}
})
}
wg.Wait()
if calls.Load() != 1 {
t.Fatal(calls.Load())
}
}
func TestMailboxDrain(t *testing.T) {
m := NewMailbox()
var wg sync.WaitGroup
wg.Go(func() {
for i := 1; i <= 10; i++ {
m.Send(i)
}
m.Close()
})
sum := 0
for {
n, ok := m.Receive()
if !ok {
break
}
sum += n
}
wg.Wait()
if sum != 55 {
t.Fatal(sum)
}
}
func TestOncePanicCountsAsDone(t *testing.T) {
var once sync.Once
calls := 0
func() { defer func() { _ = recover() }(); once.Do(func() { calls++; panic("failed") }) }()
once.Do(func() { calls++ })
if calls != 1 {
t.Fatal(calls)
}
}
func TestOnceValueRepeatsPanic(t *testing.T) {
calls := 0
f := sync.OnceValue(func() int { calls++; panic("failed") })
for range 2 {
func() {
defer func() {
if r := recover(); r != "failed" {
t.Errorf("panic=%v", r)
}
}()
f()
}()
}
if calls != 1 {
t.Fatal(calls)
}
}sync，Go 1.27.2，包约定和 API 引入版本。 ↩︎ ↩︎
sync/atomic，Go 1.27.2，类型化操作、Value 约束、顺序一致性。 ↩︎ ↩︎ ↩︎ ↩︎
bytes.Buffer.Reset及 Bytes 的别名约定。 ↩︎