数据竞争与原子操作

并发编程中，最隐蔽也最危险的 bug 就是数据竞争（Data Race）。它不会导致编译错误，运行时也不一定每次都出问题，但一旦发生就会让程序行为不可预测。本篇我们将学习什么是数据竞争、如何用竞态检测器发现它、如何用 sync/atomic 包的原子操作解决它，以及理解 Go 的内存模型和 happens-before 原则。这是写出正确并发程序的另一块基石。

一、什么是数据竞争

数据竞争是指：两个或多个 goroutine 同时访问同一个变量，且至少一个是写操作，且没有同步机制保护。

1. 一个经典的竞态例子

package main

import (
	"fmt"
	"sync"
)

func main() {
	var count int
	var wg sync.WaitGroup

	for i := 0; i < 1000; i++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			count++ // 竞态：多个 goroutine 同时读写 count
		}()
	}

	wg.Wait()
	fmt.Println("期望 1000, 实际:", count) // 可能是 998、1000、甚至更小
}

这段代码期望输出 1000，但实际运行经常小于 1000。原因是 count++ 不是原子操作，它包含三步：

1. 读取 count 的当前值。
2. 加 1。
3. 写回 count。

如果两个 goroutine 同时读到相同的值（比如都读到 5），各自加 1 写回 6，本应是 7 的结果变成了 6——丢失了一次自增。

2. 数据竞争的危害

• 结果不确定：每次运行结果可能不同，难以复现。
• 难以调试：问题可能在测试时不出现，生产环境才暴露。
• 可能更严重：数据竞争不仅导致值错误，还可能导致内存破坏、崩溃。
• 未定义行为：Go 规范明确说数据竞争是未定义行为，编译器优化可能让结果更离谱。

3. 什么情况不算数据竞争

• 多个 goroutine 同时读一个变量（没有写），不算竞争。
• 有同步机制保护（channel、Mutex、atomic），不算竞争。
• 局部变量（每个 goroutine 有自己的副本），不算竞争。

二、go build -race 和 go test -race：竞态检测器

Go 内置了强大的竞态检测器（Race Detector），基于 ThreadSanitizer。只要在构建或测试时加 -race 标志，运行时就能检测数据竞争。

1. 用法

go run -race main.go      # 运行并检测
go build -race -o app     # 构建带检测的二进制
go test -race ./...       # 测试时检测

2. 检测示例

把上面的竞态代码保存为 race.go，运行 go run -race race.go：

package main

import (
	"fmt"
	"sync"
)

func main() {
	var count int
	var wg sync.WaitGroup

	for i := 0; i < 1000; i++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			count++ // ⚠️ 竞态
		}()
	}

	wg.Wait()
	fmt.Println("count:", count)
}

运行 go run -race 会输出类似这样的报告：

==================
WARNING: DATA RACE
Read at 0x... by goroutine 7:
  main.main.func1()
      race.go:12 +0x...

Previous write at 0x... by goroutine 6:
  main.main.func1()
      race.go:12 +0x...
==================

报告会指出竞争的位置（文件名和行号）、涉及的 goroutine、读写操作，非常详细。

3. 竞态检测器的特点

• 运行时检测：只有在代码实际执行到竞争点时才能发现，所以需要充分的测试覆盖。
• 有性能开销：开启 -race 程序会慢约 5-10 倍，内存增加，不要在生产环境开启。
• 建议在 CI 中开启：测试时用 -race 是发现竞态的最佳方式。
• 不是万能的：检测不到逻辑上的同步错误，只能发现真正的数据竞争。

4. 修复竞态

修复上面例子的几种方式：

package main

import (
	"fmt"
	"sync"
	"sync/atomic"
)

func main() {
	// 方式1：用 Mutex
	var mu sync.Mutex
	var count1 int
	var wg sync.WaitGroup

	for i := 0; i < 1000; i++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			mu.Lock()
			count1++
			mu.Unlock()
		}()
	}
	wg.Wait()
	fmt.Println("Mutex:", count1)

	// 方式2：用 atomic
	var count2 int64
	for i := 0; i < 1000; i++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			atomic.AddInt64(&count2, 1)
		}()
	}
	wg.Wait()
	fmt.Println("atomic:", count2)
}

三、原子操作：sync/atomic 包

sync/atomic 包提供了底层原子操作，它们由 CPU 直接支持（如 CAS 指令），不需要锁，性能比 Mutex 高。适合简单的计数器、标志位等场景。

1. 原子操作的类型

atomic 包为 int32、int64、uint32、uint64、uintptr、Pointer 提供了原子操作。常用的有：

• AddXxx：原子加法。
• LoadXxx：原子读取。
• StoreXxx：原子写入。
• SwapXxx：原子交换。
• CompareAndSwapXxx：CAS，比较并交换。

2. atomic.AddInt64：原子加法

package main

import (
	"fmt"
	"sync"
	"sync/atomic"
)

func main() {
	var count int64
	var wg sync.WaitGroup

	for i := 0; i < 1000; i++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			atomic.AddInt64(&count, 1) // 原子自增
		}()
	}

	wg.Wait()
	fmt.Println("count:", count) // 一定是 1000
}

AddInt64 可以加任意值（包括负数做减法）：

package main

import (
	"fmt"
	"sync/atomic"
)

func main() {
	var n int64 = 100
	atomic.AddInt64(&n, 50)  // n = 150
	atomic.AddInt64(&n, -30) // n = 120
	fmt.Println(n)
}

3. atomic.LoadInt64 和 atomic.StoreInt64：原子读写

普通的 n = x 和 x = n 在并发下不安全——读可能读到写了一半的值，写也可能被读到中间状态。用 Load 和 Store 保证原子性：

package main

import (
	"fmt"
	"sync"
	"sync/atomic"
)

func main() {
	var config int64
	var wg sync.WaitGroup

	// 写入配置
	wg.Add(1)
	go func() {
		defer wg.Done()
		atomic.StoreInt64(&config, 42) // 原子写
	}()

	// 读取配置
	wg.Add(1)
	go func() {
		defer wg.Done()
		val := atomic.LoadInt64(&config) // 原子读
		fmt.Println("读到:", val)
	}()

	wg.Wait()
	fmt.Println("最终:", atomic.LoadInt64(&config))
}

为什么需要 Load/Store？对于 int64，在 32 位平台上一次读写可能不是原子的（需要两条指令）。用 atomic 保证在任何平台都是原子的。

4. atomic.CompareAndSwapInt64（CAS）

CAS 是无锁编程的核心：比较并交换。它先比较当前值是否等于期望值，相等则更新为新值，返回 true；不相等则不更新，返回 false。

package main

import (
	"fmt"
	"sync/atomic"
)

func main() {
	var value int64 = 10

	// CAS：期望 10，更新为 20
	ok := atomic.CompareAndSwapInt64(&value, 10, 20)
	fmt.Printf("CAS(10→20): ok=%v, value=%d\n", ok, value) // true, 20

	// CAS：期望 10（但现在是 20），更新为 30
	ok = atomic.CompareAndSwapInt64(&value, 10, 30)
	fmt.Printf("CAS(10→30): ok=%v, value=%d\n", ok, value) // false, 20
}

5. 用 CAS 实现自旋锁

package main

import (
	"fmt"
	"sync"
	"sync/atomic"
)

type SpinLock struct {
	flag int64
}

func (s *SpinLock) Lock() {
	// 自旋：不断尝试 CAS，直到成功
	for !atomic.CompareAndSwapInt64(&s.flag, 0, 1) {
		// 忙等
	}
}

func (s *SpinLock) Unlock() {
	atomic.StoreInt64(&s.flag, 0)
}

func main() {
	var lock SpinLock
	var count int
	var wg sync.WaitGroup

	for i := 0; i < 100; i++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			lock.Lock()
			count++
			lock.Unlock()
		}()
	}

	wg.Wait()
	fmt.Println("count:", count)
}

自旋锁在临界区很短时比 Mutex 高效（避免 goroutine 切换），但临界区长时会浪费 CPU。

6. 用 CAS 实现原子更新模式

package main

import (
	"fmt"
	"sync/atomic"
)

// 原子地把 value 更新为 f(value)
func atomicUpdate(value *int64, f func(int64) int64) {
	for {
		old := atomic.LoadInt64(value)
		new := f(old)
		if atomic.CompareAndSwapInt64(value, old, new) {
			return
		}
		// CAS 失败说明 value 被改了，重试
	}
}

func main() {
	var n int64 = 10
	atomicUpdate(&n, func(v int64) int64 { return v * 2 })
	fmt.Println(n) // 20

	atomicUpdate(&n, func(v int64) int64 { return v + 5 })
	fmt.Println(n) // 25
}

四、atomic.Value：通用原子值

atomic.Value 可以原子地存取任意类型的值（用 any 接口存储），适合「整体替换」的场景，如配置更新。

1. 基本用法

package main

import (
	"fmt"
	"sync/atomic"
)

type Config struct {
	Timeout int
	Retries int
}

var configValue atomic.Value

func main() {
	// 必须先 Store 一个初始值
	configValue.Store(&Config{Timeout: 5, Retries: 3})

	// 原子读取
	cfg := configValue.Load().(*Config)
	fmt.Printf("配置: %+v\n", cfg)

	// 原子更新（整体替换）
	configValue.Store(&Config{Timeout: 10, Retries: 5})

	cfg = configValue.Load().(*Config)
	fmt.Printf("更新后: %+v\n", cfg)
}

2. 并发安全配置更新

这是 atomic.Value 的经典场景：一个 goroutine 定期更新配置，多个 goroutine 并发读取。

package main

import (
	"fmt"
	"sync"
	"sync/atomic"
	"time"
)

type Config struct {
	Timeout int
	Hosts   []string
}

var config atomic.Value

func worker(id int, wg *sync.WaitGroup) {
	defer wg.Done()
	for i := 0; i < 3; i++ {
		cfg := config.Load().(*Config) // 并发安全读取
		fmt.Printf("worker %d: 读取配置 timeout=%d\n", id, cfg.Timeout)
		time.Sleep(300 * time.Millisecond)
	}
}

func main() {
	config.Store(&Config{Timeout: 5, Hosts: []string{"a", "b"}})

	var wg sync.WaitGroup
	// 启动多个 worker 读配置
	for i := 1; i <= 3; i++ {
		wg.Add(1)
		go worker(i, &wg)
	}

	// 后台更新配置
	go func() {
		time.Sleep(400 * time.Millisecond)
		config.Store(&Config{Timeout: 10, Hosts: []string{"a", "b", "c"}})
		fmt.Println(">>> 配置已更新")
	}()

	wg.Wait()
	fmt.Println("完成")
}

3. atomic.Value 的限制

• 类型必须一致：第一次 Store 什么类型，后续必须 Store 相同类型，否则 panic。
• 不能存 nil：Store nil 会 panic。
• 不适合频繁更新：每次 Store 都是新对象，频繁更新有 GC 压力。
• 读取的值不可变：拿到的是快照，不应修改（如果要修改，应该整体替换新对象）。

package main

import (
	"sync/atomic"
)

func main() {
	var v atomic.Value
	v.Store(42)
	// v.Store("hello") // panic：类型不一致

	// v.Store(nil) // panic：不能存 nil

	var p *int
	// v.Store(p) // panic：存了 nil 指针
	_ = p
}

五、Mutex vs atomic 性能对比

对于简单的计数器，atomic 比 Mutex 快很多。来看一个对比：

package main

import (
	"fmt"
	"sync"
	"sync/atomic"
	"time"
)

func main() {
	const N = 1000000

	// 用 atomic
	var counter1 int64
	start := time.Now()
	var wg1 sync.WaitGroup
	wg1.Add(N)
	for i := 0; i < N; i++ {
		go func() {
			defer wg1.Done()
			atomic.AddInt64(&counter1, 1)
		}()
	}
	wg1.Wait()
	atomicTime := time.Since(start)

	// 用 Mutex
	var mu sync.Mutex
	var counter2 int64
	start = time.Now()
	var wg2 sync.WaitGroup
	wg2.Add(N)
	for i := 0; i < N; i++ {
		go func() {
			defer wg2.Done()
			mu.Lock()
			counter2++
			mu.Unlock()
		}()
	}
	wg2.Wait()
	mutexTime := time.Since(start)

	fmt.Printf("atomic: %v (结果 %d)\n", atomicTime, counter1)
	fmt.Printf("Mutex:  %v (结果 %d)\n", mutexTime, counter2)
}

通常 atomic 会快 2-5 倍，因为：

• atomic 是无锁的，没有 goroutine 切换开销。
• atomic 操作是单条 CPU 指令（如 LOCK XADD），Mutex 需要进入运行时调度。

但 atomic 只适合简单场景。保护多步操作、复杂临界区，还是必须用 Mutex。

选择建议

• 单个数值的原子读写：用 atomic。
• 保护复杂状态、多步操作：用 Mutex。
• 整体对象替换：用 atomic.Value。
• 不确定时用 Mutex：它更通用，性能差异在多数场景可忽略。

六、常见数据竞争场景与修复

1. map 并发读写

package main

import (
	"fmt"
	"sync"
)

// ❌ 竞态：并发读写 map
func badMap() {
	m := make(map[int]int)
	var wg sync.WaitGroup
	for i := 0; i < 10; i++ {
		wg.Add(1)
		go func(i int) {
			defer wg.Done()
			m[i] = i // 竞态，可能 panic
		}(i)
	}
	wg.Wait()
	fmt.Println(len(m))
}

// ✅ 修复：用 sync.Map 或 Mutex+map
func goodMap() {
	var m sync.Map
	var wg sync.WaitGroup
	for i := 0; i < 10; i++ {
		wg.Add(1)
		go func(i int) {
			defer wg.Done()
			m.Store(i, i)
		}(i)
	}
	wg.Wait()
	count := 0
	m.Range(func(k, v any) bool {
		count++
		return true
	})
	fmt.Println(count)
}

func main() {
	goodMap()
}

Go 的普通 map 并发读写会直接 panic（concurrent map writes），这是运行时主动检测的。

2. 切片并发 append

package main

import (
	"fmt"
	"sync"
)

// ❌ 竞态：并发 append
func badSlice() {
	var s []int
	var wg sync.WaitGroup
	for i := 0; i < 100; i++ {
		wg.Add(1)
		go func(i int) {
			defer wg.Done()
			s = append(s, i) // 竞态，可能丢数据
		}(i)
	}
	wg.Wait()
	fmt.Println("bad:", len(s))
}

// ✅ 修复：用 Mutex 或 channel 收集
func goodSlice() {
	var mu sync.Mutex
	var s []int
	var wg sync.WaitGroup
	for i := 0; i < 100; i++ {
		wg.Add(1)
		go func(i int) {
			defer wg.Done()
			mu.Lock()
			s = append(s, i)
			mu.Unlock()
		}(i)
	}
	wg.Wait()
	fmt.Println("good:", len(s))
}

func main() {
	goodSlice()
}

3. 检查-执行（check-then-act）竞态

package main

import (
	"fmt"
	"sync"
)

// ❌ 竞态：先检查再操作
var balance int

func badWithdraw(amount int) bool {
	if balance >= amount { // 检查
		balance -= amount // 执行
		return true
	}
	return false
}

// ✅ 修复：用 Mutex 保护整个检查-执行
var mu sync.Mutex

func goodWithdraw(amount int) bool {
	mu.Lock()
	defer mu.Unlock()
	if balance >= amount {
		balance -= amount
		return true
	}
	return false
}

func main() {
	balance = 100
	var wg sync.WaitGroup
	for i := 0; i < 5; i++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			goodWithdraw(30)
		}()
	}
	wg.Wait()
	fmt.Println("余额:", balance) // 应该是 0
}

检查和执行之间如果有其他 goroutine 插入，就会出错。这种「检查-执行」必须放在同一个临界区。

4. 闭包捕获循环变量（Go 1.22 前）

package main

import (
	"fmt"
	"sync"
)

func main() {
	var wg sync.WaitGroup
	// Go 1.22+ 已修复：每次迭代 i 是独立的
	for i := 0; i < 5; i++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			fmt.Println(i) // Go 1.22 前可能都输出 5
		}()
	}
	wg.Wait()

	// 显式传参永远安全
	for i := 0; i < 5; i++ {
		wg.Add(1)
		go func(n int) {
			defer wg.Done()
			fmt.Println("safe:", n)
		}(i)
	}
	wg.Wait()
}

七、happens-before 原则

要理解为什么有些代码是安全的、有些不是，需要理解 Go 的内存模型中的 happens-before 关系。

1. 什么是 happens-before

如果操作 A 「happens-before」 操作 B，那么 A 的效果对 B 是可见的。换句话说，B 能看到 A 做的修改。如果没有 happens-before 关系，A 和 B 就是并发的，它们之间可能看到对方的中间状态。

2. Go 中的 happens-before 规则

• 单 goroutine 内：代码顺序就是 happens-before 顺序（程序顺序）。
• goroutine 创建：go 语句 happens-before 新 goroutine 的执行。
• goroutine 销毁：goroutine 退出不 happens-before 任何操作（除非用 channel 或 sync 等同步）。
• channel 发送 happens-before 对应的接收完成。
• close channel happens-before 从该 channel 收到零值。
• 无缓冲 channel 的接收 happens-before 对应的发送完成。
• Mutex Unlock happens-before 下一次 Lock。
• Once 的 Do(f) 中 f 返回 happens-before 任何其他 Do(f) 返回。

3. 为什么需要 happens-before

看这个例子：

package main

import "fmt"

var a string
var done bool

func setup() {
	a = "hello"
	done = true
}

func main() {
	go setup()
	// ❌ 没有 happens-before 关系，可能永远看不到 done = true
	for !done {
	}
	fmt.Println(a) // 可能是 "hello"，也可能是空字符串
}

主 goroutine 的 for !done 和 setup 中的 done = true 没有同步，编译器可能优化掉循环（认为 done 不会被改），或者主 goroutine 看不到 setup 的修改。这就是缺少 happens-before 的后果。

4. 用 channel 建立 happens-before

package main

import "fmt"

var a string

func main() {
	done := make(chan struct{})
	go func() {
		a = "hello"
		close(done) // close happens-before 接收
	}()
	<-done        // 接收完成，保证能看到 a = "hello"
	fmt.Println(a) // 一定是 "hello"
}

close(done) happens-before <-done，而 a = "hello" 在 close 之前（程序顺序），所以 <-done 之后的代码一定能看到 a 的值。

八、内存模型简介

Go 内存模型定义了「在一个 goroutine 中对变量的写入，在什么条件下能被另一个 goroutine 观察到」。它的核心是：没有同步的并发读写是未定义行为。

1. 核心原则

一个 goroutine 的写操作，只有通过同步机制（channel、Mutex、atomic、Once 等）建立 happens-before 关系后，才能被另一个 goroutine 可靠地观察到。

2. 不要依赖「直觉」

package main

import "fmt"

var x, y int

func main() {
	// 假设一个 goroutine 写 x，另一个读 x
	// 没有同步的话，读者可能读到 0（旧值），也可能读到新值
	// 甚至在重排序优化下读到不一致的状态
	go func() {
		x = 1
		fmt.Println("写完 x")
	}()
	fmt.Println("读 x:", x) // 不保证读到 1
}

3. 正确的同步方式

package main

import "fmt"

var x int

func main() {
	ch := make(chan int)
	go func() {
		x = 1
		ch <- x // 发送 happens-before 接收
	}()
	v := <-ch       // 接收完成，保证看到 x = 1
	fmt.Println(v, x) // 1 1
}

4. atomic 也建立 happens-before

package main

import (
	"fmt"
	"sync/atomic"
)

var (
	data    int
	ready   int64
)

func main() {
	go func() {
		data = 42                    // 写数据
		atomic.StoreInt64(&ready, 1) // 标记就绪（happens-before 之后的 Load）
	}()

	for atomic.LoadInt64(&ready) == 0 {
		// 等待 ready
	}
	// Load 看到 ready=1，则之前的 Store 已经 happens-before
	// 所以一定能看到 data = 42
	fmt.Println("data:", data) // 42
}

atomic.Store 和 atomic.Load 之间有 happens-before 关系，所以 data = 42 对主 goroutine 可见。这是实现「发布模式」的标准做法。

九、实战：用 atomic 实现安全的单例

package main

import (
	"fmt"
	"sync"
	"sync/atomic"
)

type Singleton struct{ name string }

var (
	instance *Singleton
	once     sync.Once
)

// 用 Once（最推荐）
func GetInstance() *Singleton {
	once.Do(func() {
		instance = &Singleton{name: "单例"}
	})
	return instance
}

// 用 atomic.Value 也能实现
var atomicInstance atomic.Value

func GetInstanceAtomic() *Singleton {
	v := atomicInstance.Load()
	if v == nil {
		// 首次需要初始化（这里简化，实际要防并发初始化，用 Once 更好）
		atomicInstance.Store(&Singleton{name: "原子单例"})
	}
	return atomicInstance.Load().(*Singleton)
}

func main() {
	s1 := GetInstance()
	s2 := GetInstance()
	fmt.Println(s1 == s2) // true

	s3 := GetInstanceAtomic()
	fmt.Println(s3.name)
}

对于单例模式，sync.Once 是最简单正确的方式，atomic.Value 适合「可变配置」场景。

十、小结

本篇我们学习了数据竞争和原子操作：

1. 数据竞争：多个 goroutine 同时访问同一变量，至少一个写，无同步保护。危害是结果不确定、难以复现、未定义行为。count++、map 读写、check-then-act 都是常见竞态。

2. 竞态检测器：go run/test/build -race 能在运行时检测数据竞争。建议在 CI 中测试时开启。有性能开销，不要在生产开启。只能发现实际执行到的竞态，需要充分测试覆盖。

3. atomic 包：AddXxx 原子加法、LoadXxx 原子读、StoreXxx 原子写、SwapXxx 原子交换、CompareAndSwapXxx CAS 比较并交换。适合简单计数器和标志位，性能比 Mutex 高 2-5 倍。

4. CAS 应用：自旋锁、原子更新模式（load → compute → CAS → retry）。是无锁编程的核心。

5. atomic.Value：原子存取任意类型值，适合配置整体替换、发布模式。类型必须一致、不能存 nil、读取的值不可变。

6. Mutex vs atomic：单数值原子读写用 atomic，复杂临界区用 Mutex，对象整体替换用 atomic.Value，不确定用 Mutex。

7. 常见竞态修复：map 并发用 sync.Map 或 Mutex+map；切片 append 用 Mutex；check-then-act 整体放临界区；循环变量显式传参。

8. happens-before：A happens-before B 表示 A 的效果对 B 可见。Go 有明确的规则：channel 发送 happens-before 接收、close happens-before 收零值、Mutex Unlock happens-before 下次 Lock、Once.Do 的 f 返回 happens-before 其他 Do 返回等。

9. 内存模型：没有同步的并发读写是未定义行为。必须通过 channel、Mutex、atomic、Once 建立同步。atomic 的 Store/Load 也建立 happens-before，可用于「发布模式」。

下一篇我们将系统总结并发编程中的各种陷阱和最佳实践，把这些知识串联起来。