并发编程是每一个后端工程师的必修课而锁是并发编程中最核心也最容易翻车的概念。本文将从 CPU 指令级别一直讲到分布式系统带你彻底搞懂并发锁。一、并发问题的本质从一行i讲起看这段看似无害的代码# 两个线程同时执行counter0defincrement():globalcounterfor_inrange(100000):counter1两个线程各加 10 万次你期望结果是 20 万。跑一下importthreading counter0defincrement():globalcounterfor_inrange(100000):counter1t1threading.Thread(targetincrement)t2threading.Thread(targetincrement)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 R1100 100 T2 LOAD R2100 100 T3 ADD R1101 100 T4 STORE 101 101 T5 ADD R2101 101 T6 STORE 101 101 ← 丢了一次更新这就是竞态条件Race Condition——多个线程同时访问共享数据最终结果取决于线程调度时序不可预测。二、锁的本质把并行变串行2.1 互斥锁Mutex互斥锁是最基本的锁同一时刻只允许一个线程进入临界区。importthreading lockthreading.Lock()counter0defincrement():globalcounterfor_inrange(100000):withlock:# 获取锁counter1# 临界区# 自动释放锁# 这次一定是 200000工作方式Thread-1: 获取锁 → 执行临界区 → 释放锁 Thread-2: 等待...等待... → 获取锁 → 执行临界区 → 释放锁底层实现依赖于 CPU 提供的原子指令比如 x86 的CMPXCHGCompare-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 自旋锁特性互斥锁 (Mutex)自旋锁 (Spinlock)等待方式线程挂起、让出 CPUCPU 空转循环检查适用场景临界区执行时间长临界区极短几行代码上下文切换有开销大无CPU 消耗低高空转浪费用户态/内核态需要内核参与纯用户态// Go 中的互斥锁varmu sync.Mutex mu.Lock()// 临界区mu.Unlock()// Go 中的自旋runtime 内部使用// 用户代码不直接暴露自旋锁但 sync.Mutex 内部会短暂自旋后回退到信号量选型经验临界区超过几次函数调用就该用 Mutex只是改个标志位、加减个计数器Spinlock 更合适。三、进阶锁类型3.1 读写锁RWMutex读多写少的场景下互斥锁太暴力了——读操作之间明明不冲突却被强行串行化。varrw sync.RWMutexvarcachemake(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 个线程同时执行semthreading.Semaphore(3)defworker(name):withsem:print(f{name}进入)time.sleep(1)# 模拟工作print(f{name}离开)# 5 个线程但最多 3 个同时运行foriinrange(5):threading.Thread(targetworker,args(fT{i},)).start()输出T0 进入 T1 进入 T2 进入 T3、T4 等待 T0 离开 → T3 进入 T1 离开 → T4 进入典型场景数据库连接池、限流器、生产者-消费者模型。3.3 条件变量Condition Variable条件变量解决的是等某个条件满足再继续的问题importthreading condthreading.Condition()items[]defproducer():foriinrange(5):withcond:items.append(i)print(f生产{i})cond.notify()# 通知等待的消费者defconsumer():for_inrange(5):withcond:whilenotitems:# 用 while 而非 ifcond.wait()# 释放锁 等待通知itemitems.pop(0)print(f消费{item})# 同时启动threading.Thread(targetconsumer).start()threading.Thread(targetproducer).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()}死锁四要素所有死锁都满足互斥资源不能共享持有并等待线程持有锁的同时等待其他锁不可抢占锁不能被外力剥夺循环等待线程间形成等待环路 A→B→C→A4.2 预防死锁的四条法则法则 1 — 固定加锁顺序所有线程按相同顺序获取锁。// 定义锁的层级const(LOCK_LEVEL_DB1// 最内层LOCK_LEVEL_CACHE2LOCK_LEVEL_LOG3// 最外层)// 永远从高层到低层加锁logMu.Lock()// L3cacheMu.Lock()// L2dbMu.Lock()// L1// 释放时从低层到高层dbMu.Unlock()cacheMu.Unlock()logMu.Unlock()法则 2 — 超时放弃不要无限等待。iflock.acquire(timeout5):try:# 临界区finally:lock.release()else:# 5 秒没拿到放弃并重试整个操作raiseTimeoutError(获取锁超时请重试)法则 3 — tryLock 探测先试试能不能一次拿完所有锁拿不到就全放。funcTryBothLocks()bool{if!mu1.TryLock(){returnfalse}if!mu2.TryLock(){mu1.Unlock()// 拿不到第二个就释放第一个returnfalse}returntrue}法则 4 — 减少锁的粒度大锁拆小锁降低锁冲突概率。// 坏一把大锁classBadCache{privatefinalMapString,StringmapnewHashMap();publicsynchronizedStringget(Stringk){returnmap.get(k);}publicsynchronizedvoidput(Stringk,Stringv){map.put(k,v);}}// 好分段锁classGoodCacheK,V{privatefinalintSEGMENTS16;privatefinalReentrantLock[]locksnewReentrantLock[SEGMENTS];privatefinalMapK,V[]segmentsnewHashMap[SEGMENTS];publicVget(Kkey){intidxMath.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.balanceamount to.mu.Unlock()from.mu.Unlock()return}from.mu.Unlock()}// 关键加随机退避避免同步重试time.Sleep(time.Duration(rand.Intn(100))*time.Millisecond)}}不加随机退避两个转账同时进行就会活锁。加了随机退避其中一个会先拿到两个锁。六、无锁编程CAS 与原子操作锁的问题是开销大、容易死锁。无锁Lock-Free编程用原子操作替代锁。6.1 CASCompare-And-Swap# Python 3.13 可以用 ctypes 调用系统 CAS# 用 Go 来演示更直观importsync/atomicvar counter int64//不用锁的并发安全递增 func increment(){for{old:atomic.LoadInt64(counter)new:old1ifatomic.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: 读到 A1准备 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 提供了AtomicStampedReferencePython 可以用tuple版本号 不可变对象。七、语言选择指南场景推荐方案理由Python 多线程threading.Lock 避免 CPU 密集GIL 限制多线程适合 IO锁开销低Python 多进程multiprocessing.Lock跨进程基于共享内存Go 高并发sync.Mutex/changoroutine 轻量用 channel 通信优于共享内存Java 企业级ReentrantLockCondition功能最全支持公平锁、可中断锁Rust 系统级std::sync::MutexT编译期所有权检查根本上防止数据竞争Redis 分布式SETNX Lua 脚本简单可靠适合分布式环境数据库并发SELECT ... FOR UPDATE利用数据库行锁适合有 DB 的场景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 rredis.Redis()# 获取锁ifr.set(lock:order:123,unique_id,nxTrue,ex10):try:# 处理订单 123process_order(123)finally:# 释放锁——必须验证是不是自己持有的ifr.get(lock:order:123)unique_id:r.delete(lock:order:123)三个必须注意的点要设过期时间ex10防止持有者挂了锁永远不释放释放时要验证持有者身份防止误删别人的锁过期时间要足够长不能比业务执行时间短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 数据库乐观锁-- 版本号方式UPDATEproductsSETstockstock-1,versionversion1WHEREid123ANDstock0ANDversion5;-- 如果 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 endredisClient.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.RowsAffected0{returnerrors.New(抢光了)}// 创建订单...returnnil}设计思路多层漏斗越外层拦截率越高越内层越精确。本地内存 → 分布式锁 → 数据库行锁层层过滤保护下游。十、总结锁的选择决策树需要跨进程/跨服务器 ├── 是 → 分布式锁 │ ├── 强一致性要求 → Redlock / 数据库悲观锁 │ └── 允许短暂不一致 → Redis SETNX / 数据库乐观锁 └── 否单机→ 本地锁 ├── 临界区极短几行代码→ 自旋锁 / 原子操作 ├── 临界区稍长 → Mutex ├── 读多写少 → RWMutex读写锁 ├── 需要限流/控制并发数 → Semaphore信号量 └── 需要等待条件 → Condition Variable条件变量最后三条铁律锁的粒度要小——锁住的代码越少越好把 IO 操作移出临界区加锁顺序要统一——永远用同一个顺序死锁就不会发生能不用锁就不用锁——immutable 数据、channel 通信、原子操作很多时候可以完全避免加锁并发编程的本质不是怎样用锁而是怎样不用锁还能保证正确性。最好的锁是没有锁。