atomic 原子操作真的“原子“吗？CPU 指令真相解析-编程阁

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) }