并发编程是每一个后端工程师的必修课，而锁是并发编程中最核心也最容易翻车的概念。本文将从 CPU 指令级别一直讲到分布式系统，带你彻底搞懂并发锁。

一、并发问题的本质：从一行i++讲起

看这段看似无害的代码：

# 两个线程同时执行counter=0defincrement():globalcounterfor_inrange(100000):counter+=1

两个线程各加 10 万次，你期望结果是 20 万。跑一下：

importthreading counter=0defincrement():globalcounterfor_inrange(100000):counter+=1t1=threading.Thread(target=increment)t2=threading.Thread(target=increment)t1.start();t2.start()t1.join();t2.join()print(counter)# 输出: 143287 ？？？

为什么不是 200000？因为counter += 1在 CPU 层面不是原子操作，它经历了三步：

LOAD R1, [counter] # 从内存读取 counter 到寄存器 ADD R1, 1 # 寄存器加 1 STORE [counter], R1 # 写回内存

两个线程交错执行时：

时间轴 Thread-1 Thread-2 counter(内存) ────────────────────────────────────────────────────── T1 LOAD R1=100 100 T2 LOAD R2=100 100 T3 ADD R1=101 100 T4 STORE 101 101 T5 ADD R2=101 101 T6 STORE 101 101 ← 丢了一次更新！

这就是竞态条件（Race Condition）——多个线程同时访问共享数据，最终结果取决于线程调度时序，不可预测。

二、锁的本质：把并行变串行

2.1 互斥锁（Mutex）

互斥锁是最基本的锁：同一时刻只允许一个线程进入临界区。

importthreading lock=threading.Lock()counter=0defincrement():globalcounterfor_inrange(100000):withlock:# 获取锁counter+=1# 临界区# 自动释放锁# 这次一定是 200000

工作方式：

Thread-1: 获取锁 → 执行临界区 → 释放锁 Thread-2: 等待...等待... → 获取锁 → 执行临界区 → 释放锁

底层实现依赖于 CPU 提供的原子指令，比如 x86 的CMPXCHG（Compare-and-Swap）：

// 自旋锁的伪实现typedefstruct{intlocked;// 0=空闲, 1=已占用}spinlock_t;voidspin_lock(spinlock_t*lock){while(1){// 原子操作：如果 lock->locked == 0，则设为 1 并返回旧值 0if(__sync_bool_compare_and_swap(&lock->locked,0,1))break;// 获取成功// 否则自旋等待}}voidspin_unlock(spinlock_t*lock){__sync_lock_release(&lock->locked);// 设为 0}

2.2 互斥锁 vs 自旋锁

// Go 中的互斥锁varmu sync.Mutex mu.Lock()// 临界区mu.Unlock()// Go 中的自旋（runtime 内部使用）// 用户代码不直接暴露自旋锁，但 sync.Mutex 内部会短暂自旋后回退到信号量

选型经验：临界区超过几次函数调用就该用 Mutex；只是改个标志位、加减个计数器，Spinlock 更合适。

三、进阶锁类型

3.1 读写锁（RWMutex）

读多写少的场景下，互斥锁太"暴力"了——读操作之间明明不冲突，却被强行串行化。

varrw sync.RWMutexvarcache=make(map[string]string)// 多个 goroutine 可以同时读funcGet(keystring)string{rw.RLock()// 读锁，不互斥deferrw.RUnlock()returncache[key]}// 写操作独占funcSet(key,valuestring){rw.Lock()// 写锁，互斥deferrw.Unlock()cache[key]=value}

读写锁的状态机：

┌──────────┐ │ 空闲 │ └────┬─────┘ ┌──────┴──────┐ ▼ ▼ ┌─────────┐ ┌──────────┐ │ 读锁 │ │ 写锁 │ │ (多持) │ │ (独占) │ └────┬────┘ └────┬─────┘ │ │ 读锁=0 释放 释放 ▼ ▼ 空闲状态 空闲状态

注意：读写锁有写者饥饿问题——如果一直有新的读者进来，写者可能永远拿不到锁。Linux 内核的rwlock通过"写者优先"策略解决。

3.2 信号量（Semaphore）

信号量控制的是"同时允许 N 个线程访问"，而不是"只允许 1 个"。

importthreading# 信号量值为 3：最多允许 3 个线程同时执行sem=threading.Semaphore(3)defworker(name):withsem:print(f"{name}进入")time.sleep(1)# 模拟工作print(f"{name}离开")# 5 个线程，但最多 3 个同时运行foriinrange(5):threading.Thread(target=worker,args=(f"T{i}",)).start()

输出：

T0 进入 T1 进入 T2 进入 （T3、T4 等待） T0 离开 → T3 进入 T1 离开 → T4 进入

典型场景：数据库连接池、限流器、生产者-消费者模型。

3.3 条件变量（Condition Variable）

条件变量解决的是"等某个条件满足再继续"的问题：

importthreading cond=threading.Condition()items=[]defproducer():foriinrange(5):withcond:items.append(i)print(f"生产{i}")cond.notify()# 通知等待的消费者defconsumer():for_inrange(5):withcond:whilenotitems:# 用 while 而非 if！cond.wait()# 释放锁 + 等待通知item=items.pop(0)print(f"消费{item}")# 同时启动threading.Thread(target=consumer).start()threading.Thread(target=producer).start()

为什么用while而不是if？因为虚假唤醒（Spurious Wakeup）——线程可能在没有notify()的情况下被唤醒。while保证醒来后重新检查条件。

四、死锁：并发编程的头号杀手

4.1 经典死锁场景

varmu1,mu2 sync.Mutex// goroutine AfuncA(){mu1.Lock()time.Sleep(100*time.Millisecond)// 模拟工作mu2.Lock()// 等 mu2// ...mu2.Unlock()mu1.Unlock()}// goroutine BfuncB(){mu2.Lock()time.Sleep(100*time.Millisecond)mu1.Lock()// 等 mu1 ← 互相等待，死锁！// ...mu1.Unlock()mu2.Unlock()}

死锁四要素（所有死锁都满足）：

1. 互斥：资源不能共享
2. 持有并等待：线程持有锁的同时等待其他锁
3. 不可抢占：锁不能被外力剥夺
4. 循环等待：线程间形成等待环路 A→B→C→A

4.2 预防死锁的四条法则

法则 1 — 固定加锁顺序：所有线程按相同顺序获取锁。

// 定义锁的层级const(LOCK_LEVEL_DB=1// 最内层LOCK_LEVEL_CACHE=2LOCK_LEVEL_LOG=3// 最外层)// 永远从高层到低层加锁logMu.Lock()// L3cacheMu.Lock()// L2dbMu.Lock()// L1// 释放时从低层到高层dbMu.Unlock()cacheMu.Unlock()logMu.Unlock()

法则 2 — 超时放弃：不要无限等待。

iflock.acquire(timeout=5):try:# 临界区finally:lock.release()else:# 5 秒没拿到，放弃并重试整个操作raiseTimeoutError("获取锁超时，请重试")

法则 3 — tryLock 探测：先试试能不能一次拿完所有锁，拿不到就全放。

funcTryBothLocks()bool{if!mu1.TryLock(){returnfalse}if!mu2.TryLock(){mu1.Unlock()// 拿不到第二个就释放第一个returnfalse}returntrue}

法则 4 — 减少锁的粒度：大锁拆小锁，降低锁冲突概率。

// 坏：一把大锁classBadCache{privatefinalMap<String,String>map=newHashMap<>();publicsynchronizedStringget(Stringk){returnmap.get(k);}publicsynchronizedvoidput(Stringk,Stringv){map.put(k,v);}}// 好：分段锁classGoodCache<K,V>{privatefinalintSEGMENTS=16;privatefinalReentrantLock[]locks=newReentrantLock[SEGMENTS];privatefinalMap<K,V>[]segments=newHashMap[SEGMENTS];publicVget(Kkey){intidx=Math.abs(key.hashCode()%SEGMENTS);locks[idx].lock();try{returnsegments[idx].get(key);}finally{locks[idx].unlock();}}}

五、活锁与饥饿

死锁是"大家都死了"，活锁是"大家都在动，但谁也办不成事"：

走廊里两个人相向而行： A 往左让 → B 也往左让 → A 往右让 → B 也往右让 → ... 无限循环，谁也没过去

代码示例：

functransfer(from,to*Account,amountint){for{iffrom.mu.TryLock(){ifto.mu.TryLock(){// 两个锁都拿到了，执行转账from.balance-=amount to.balance+=amount to.mu.Unlock()from.mu.Unlock()return}from.mu.Unlock()}// 关键：加随机退避，避免同步重试time.Sleep(time.Duration(rand.Intn(100))*time.Millisecond)}}

不加随机退避，两个转账同时进行就会活锁。加了随机退避，其中一个会先拿到两个锁。

六、无锁编程：CAS 与原子操作

锁的问题是开销大、容易死锁。无锁（Lock-Free）编程用原子操作替代锁。

6.1 CAS（Compare-And-Swap）

# Python 3.13+ 可以用 ctypes 调用系统 CAS# 用 Go 来演示更直观import"sync/atomic"var counter int64//不用锁的并发安全递增 func increment(){for{old:=atomic.LoadInt64(&counter)new:=old+1ifatomic.CompareAndSwapInt64(&counter,old,new){break//CAS 成功}//CAS 失败说明被其他 goroutine 抢先了，重试}}

CAS 的工作流程：

1. 读取当前值: old = *addr 2. 计算新值: new = old + 1 3. 原子比较并交换: if *addr == old { *addr = new; return true } else { return false } ← 失败了，重试

6.2 ABA 问题

CAS 有个经典陷阱——ABA 问题：

初始值 A = 1 Thread-1: 读到 A=1，准备 CAS(1, 2) Thread-2: CAS(1, 3) 成功，A 变成 3 Thread-2: CAS(3, 1) 成功，A 变回 1 Thread-1: CAS(1, 2) 成功 ← 其实中间已经变过两次了！

解决：加版本号。

// Go 的 atomic.Value 或使用带版本号的 CAStypeVersionedValuestruct{valueint64versionint64}func(v*VersionedValue)CAS(oldVal,newValint64,oldVer,newVerint64)bool{returnatomic.CompareAndSwapInt64(&v.value,oldVal,newVal)&&atomic.CompareAndSwapInt64(&v.version,oldVer,newVer)}

Java 提供了AtomicStampedReference，Python 可以用tuple版本号 + 不可变对象。

七、语言选择指南

Go 的并发哲学：用通信共享内存

// 不推荐：用锁共享内存varmu sync.Mutexvardatamap[string]int// 推荐：用 channel 传递数据的所有权ch:=make(chanmap[string]int)gofunc(){data:=make(map[string]int)data["key"]=42ch<-data// 所有权转移给接收方}()received:=<-ch// 现在你是 data 的唯一拥有者

八、分布式锁

单机锁管不了多台服务器的并发。分布式锁登场。

8.1 Redis 分布式锁

最简单的实现（有坑）：

importredis r=redis.Redis()# 获取锁ifr.set("lock:order:123","unique_id",nx=True,ex=10):try:# 处理订单 123process_order(123)finally:# 释放锁——必须验证是不是自己持有的！ifr.get("lock:order:123")=="unique_id":r.delete("lock:order:123")

三个必须注意的点：

1. 要设过期时间（ex=10），防止持有者挂了锁永远不释放
2. 释放时要验证持有者身份，防止误删别人的锁
3. 过期时间要足够长，不能比业务执行时间短

Redlock 算法（Redis 官方推荐）：

1. 获取当前时间 T1 2. 依次向 N 个独立的 Redis 节点请求锁（SET key value NX PX ttl） 3. 计算获取锁的总耗时 = T2 - T1 4. 如果 (获取成功的节点数 >= N/2 + 1) 且 (总耗时 < TTL)： 锁获取成功，有效时间 = TTL - 总耗时 5. 如果获取失败，向所有节点发送释放请求

8.2 数据库乐观锁

-- 版本号方式UPDATEproductsSETstock=stock-1,version=version+1WHEREid=123ANDstock>0ANDversion=5;-- 如果 affected_rows = 0，说明并发冲突，重试

乐观锁适合冲突少的场景（如读多写少），不需要真正加锁，靠版本号检测冲突。

九、实战案例：秒杀系统

综合运用上面的知识，设计一个秒杀系统的锁策略：

// 秒杀扣库存——多层防护funcSeckill(productIDstring,userIDstring)error{// 第 0 层：本地内存预检（最快，拦截 90% 无效请求）if!localStock.Has(productID){returnerrors.New("已售罄")}// 第 1 层：单机互斥锁（防止同机器内并发）mu.Lock()defermu.Unlock()// 第 2 层：Redis 分布式锁（防止多机并发）lockKey:=fmt.Sprintf("seckill:lock:%s",productID)lockValue:=fmt.Sprintf("%s:%d",userID,time.Now().UnixNano())ok,err:=redisClient.SetNX(ctx,lockKey,lockValue,2*time.Second).Result()iferr!=nil||!ok{returnerrors.New("系统繁忙，请重试")}deferfunc(){// Lua 脚本保证原子释放script:=`if redis.call("get", KEYS[1]) == ARGV[1] then return redis.call("del", KEYS[1]) else return 0 end`redisClient.Eval(ctx,script,[]string{lockKey},lockValue)}()// 第 3 层：数据库乐观锁（最后一道防线）result:=db.Exec("UPDATE products SET stock = stock - 1, version = version + 1 "+"WHERE id = ? AND stock > 0 AND version = ?",productID,currentVersion,)ifresult.RowsAffected==0{returnerrors.New("抢光了")}// 创建订单...returnnil}

设计思路：多层漏斗，越外层拦截率越高，越内层越精确。本地内存 → 分布式锁 → 数据库行锁，层层过滤，保护下游。

十、总结：锁的选择决策树

需要跨进程/跨服务器？ ├── 是 → 分布式锁 │ ├── 强一致性要求 → Redlock / 数据库悲观锁 │ └── 允许短暂不一致 → Redis SETNX / 数据库乐观锁 └── 否（单机）→ 本地锁 ├── 临界区极短（几行代码）→ 自旋锁 / 原子操作 ├── 临界区稍长 → Mutex ├── 读多写少 → RWMutex（读写锁） ├── 需要限流/控制并发数 → Semaphore（信号量） └── 需要等待条件 → Condition Variable（条件变量）

最后三条铁律：

1. 锁的粒度要小——锁住的代码越少越好，把 IO 操作移出临界区
2. 加锁顺序要统一——永远用同一个顺序，死锁就不会发生
3. 能不用锁就不用锁——immutable 数据、channel 通信、原子操作，很多时候可以完全避免加锁

并发编程的本质不是"怎样用锁"，而是"怎样不用锁还能保证正确性"。

最好的锁，是没有锁。