atomic 原子操作真的"原子"吗？CPU 指令真相解析
前言
atomic 包是 Go 并发编程的利器，速度快、无锁、不阻塞。但很多人不知道，atomic 操作在底层是怎么实现的。
atomic 操作真的"原子"吗？本文研究了 CPU 指令后发现，有些操作是真正的原子，有些只是"看起来原子"。今天来聊聊。
一、 底层原理
1.1 atomic 的底层实现
atomic 操作依赖 CPU 的原子指令：
graph TD A["atomic.AddInt64"] --> B["CPU 指令"] B --> C["LOCK 前缀"] C --> D["内存总线锁定"] D --> E["原子操作"] E --> F["返回新值"] G["CAS"] --> H["比较并交换"] H --> I["原子指令"] I --> J["成功或失败"]关键点：
- LOCK 前缀确保指令原子性
- 内存总线锁定影响其他 CPU
- CAS 是无锁编程的基础
- 不同 CPU 架构实现不同
1.2 atomic 操作分类
| 操作 | 指令 | 原子性 | 性能 |
|---|---|---|---|
| AddInt64 | LOCK XADD | 真原子 | 快 |
| LoadInt64 | LOCK MOV | 真原子 | 快 |
| StoreInt64 | LOCK MOV | 真原子 | 快 |
| CompareAndSwap | LOCK CMPXCHG | 真原子 | 中 |
| Swap | LOCK XCHG | 真原子 | 快 |
二、 快速上手
2.1 atomic 的基本使用
package main import ( "fmt" "sync" "sync/atomic" ) var count int64 func main() { var wg sync.WaitGroup for i := 0; i < 100; i++ { wg.Add(1) go func() { defer wg.Done() for j := 0; j < 100000; j++ { atomic.AddInt64(&count, 1) } }() } wg.Wait() fmt.Printf("count: %d\n", count) }2.2 atomic 与 Mutex 对比
import ( "sync" "sync/atomic" "time" ) var ( atomicCount int64 mutexCount int64 mu sync.Mutex ) func atomicInc() { atomic.AddInt64(&atomicCount, 1) } func mutexInc() { mu.Lock() mutexCount++ mu.Unlock() } func main() { n := 10000000 start := time.Now() for i := 0; i < n; i++ { atomicInc() } fmt.Printf("atomic: %v\n", time.Since(start)) start = time.Now() for i := 0; i < n; i++ { mutexInc() } fmt.Printf("mutex: %v\n", time.Since(start)) }三、 核心 API / 深水区
3.1 atomic 操作速查
| 函数 | 功能 | 注意事项 |
|---|---|---|
| AddInt64 | 原子加 | 对齐要求 |
| LoadInt64 | 原子读 | 防止乱序 |
| StoreInt64 | 原子写 | 防止乱序 |
| CompareAndSwap | CAS | 乐观锁 |
| Swap | 交换 | 返回旧值 |
3.2 内存对齐要求
// 错误：没有对齐 type BadCounter struct { flag byte count int64 // 可能没对齐 } // 正确：保证对齐 type GoodCounter struct { count int64 // 8 字节对齐 flag byte }3.3 CAS 乐观锁实现
var value int64 func update(newValue int64) bool { for { old := atomic.LoadInt64(&value) if atomic.CompareAndSwapInt64(&value, old, newValue) { return true } // 失败，重试 } }四、 实战演练
4.1 高性能计数器
package main import ( "fmt" "sync" "sync/atomic" "time" ) type AtomicCounter struct { count int64 } func (c *AtomicCounter) Inc() { atomic.AddInt64(&c.count, 1) } func (c *AtomicCounter) Dec() { atomic.AddInt64(&c.count, -1) } func (c *AtomicCounter) Value() int64 { return atomic.LoadInt64(&c.count) } func main() { counter := &AtomicCounter{} var wg sync.WaitGroup start := time.Now() for i := 0; i < 100; i++ { wg.Add(1) go func() { defer wg.Done() for j := 0; j < 100000; j++ { counter.Inc() } }() } wg.Wait() fmt.Printf("耗时: %v, count: %d\n", time.Since(start), counter.Value()) }五、 避坑指南与最佳实践
💡技巧：读操作也用 Load
不加 Load 可能读到旧值，因为 CPU 乱序执行。
⚠️警告：atomic 不能替代 Mutex
复杂临界区，还得上 Mutex。
✅推荐：简单计数器用 atomic
高频场景，atomic 是你的朋友。
六、 综合实战演示
6.1 生产级并发限流器
package main import ( "fmt" "sync" "sync/atomic" "time" ) type RateLimiter struct { limit int64 current int64 interval time.Duration lastTime time.Time } func NewRateLimiter(limit int64, interval time.Duration) *RateLimiter { return &RateLimiter{ limit: limit, interval: interval, lastTime: time.Now(), } } func (rl *RateLimiter) Allow() bool { now := time.Now() if now.Sub(rl.lastTime) >= rl.interval { atomic.StoreInt64(&rl.current, 0) rl.lastTime = now } current := atomic.AddInt64(&rl.current, 1) return current <= rl.limit } func main() { rl := NewRateLimiter(100, time.Second) var wg sync.WaitGroup passed := int64(0) blocked := int64(0) for i := 0; i < 200; i++ { wg.Add(1) go func() { defer wg.Done() if rl.Allow() { atomic.AddInt64(&passed, 1) } else { atomic.AddInt64(&blocked, 1) } }() } wg.Wait() fmt.Printf("通过: %d, 阻塞: %d\n", passed, blocked) }总结
atomic 的核心要点：
- 依赖 CPU 原子指令
- 内存对齐很重要
- CAS 是无锁编程基础
- 读操作也要用 Load