一、为什么需要分布式锁单体应用中synchronized 或 ReentrantLock 就能解决并发问题。但在分布式系统中多个进程运行在不同机器上JVM 级别的锁无能为力问题场景秒杀扣库存 ┌──────────┐ ┌──────────┐ ┌──────────┐ │ 服务A │ │ 服务B │ │ 服务C │ │ (北京) │ │ (上海) │ │ (深圳) │ ├──────────┤ ├──────────┤ ├──────────┤ │ 读库存: 1 │ │ 读库存: 1 │ │ 读库存: 1 │ │ 扣减: 0 │ │ 扣减: 0 │ │ 扣减: 0 │ └──────────┘ └──────────┘ └──────────┘ 结果库存 1 件商品被卖了 3 次分布式锁的核心要求互斥性同一时刻只有一个客户端持有锁安全性不会产生死锁持有锁的客户端崩溃后能自动释放可重入同一个客户端可以多次获取同一把锁容错性锁服务自身高可用二、基于 Redis 的分布式锁2.1 基础版SETNX EXPIRE// distributed/lock/redis_lock.go package lock import ( context fmt log math/rand sync time github.com/go-redis/redis/v8 ) // RedisLock Redis 分布式锁 type RedisLock struct { client *redis.Client key string value string // 唯一标识用于安全解锁 ttl time.Duration mu sync.Mutex isLocked bool watchDog chan struct{} // 看门狗停止信号 } // NewRedisLock 创建 Redis 锁 func NewRedisLock(client *redis.Client, key string, ttl time.Duration) *RedisLock { return RedisLock{ client: client, key: key, value: generateValue(), ttl: ttl, watchDog: make(chan struct{}, 1), } } // Lock 获取锁阻塞直到成功 func (l *RedisLock) Lock(ctx context.Context) error { for { select { case -ctx.Done(): return ctx.Err() default: ok, err : l.TryLock(ctx) if err ! nil { return err } if ok { return nil } // 重试等待 time.Sleep(time.Duration(50rand.Intn(50)) * time.Millisecond) } } } // TryLock 尝试获取锁非阻塞 func (l *RedisLock) TryLock(ctx context.Context) (bool, error) { l.mu.Lock() defer l.mu.Unlock() // SET key value NX PX ttl // NX: 只有 key 不存在时才设置 // PX: 设置过期时间毫秒 ok, err : l.client.SetNX(ctx, l.key, l.value, l.ttl).Result() if err ! nil { return false, fmt.Errorf(redis setnx: %w, err) } if ok { l.isLocked true // 启动看门狗自动续期 l.startWatchDog(ctx) log.Printf([Lock] Acquired lock: %s (value%s), l.key, l.value) } return ok, nil } // Unlock 释放锁使用 Lua 脚本保证原子性 func (l *RedisLock) Unlock(ctx context.Context) error { l.mu.Lock() defer l.mu.Unlock() if !l.isLocked { return nil } // Lua 脚本检查 value 是否匹配匹配才删除 script : if redis.call(get, KEYS[1]) ARGV[1] then return redis.call(del, KEYS[1]) else return 0 end result, err : l.client.Eval(ctx, script, []string{l.key}, l.value).Result() if err ! nil { return fmt.Errorf(redis unlock eval: %w, err) } // 停止看门狗 close(l.watchDog) l.isLocked false if result.(int64) 1 { log.Printf([Lock] Released lock: %s, l.key) } else { log.Printf([Lock] Lock already expired or owned by another: %s, l.key) } return nil } // startWatchDog 启动看门狗自动续期 func (l *RedisLock) startWatchDog(ctx context.Context) { go func() { ticker : time.NewTicker(l.ttl / 3) // 每 1/3 TTL 续期一次 defer ticker.Stop() for { select { case -ticker.C: l.mu.Lock() if !l.isLocked { l.mu.Unlock() return } // 续期 _, err : l.client.Expire(ctx, l.key, l.ttl).Result() if err ! nil { log.Printf([Lock] Watchdog renew failed: %v, err) } l.mu.Unlock() case -l.watchDog: return } } }() } // generateValue 生成唯一值 func generateValue() string { return fmt.Sprintf(%d-%d, time.Now().UnixNano(), rand.Int63()) }2.2 Redlock 算法Redis 官方推荐的分布式锁算法解决单点故障问题// distributed/lock/redlock.go package lock import ( context fmt log math/rand time github.com/go-redis/redis/v8 ) // Redlock Redlock 算法实现 type Redlock struct { clients []*redis.Client // 多个 Redis 实例 quorum int // 法定数量多数派 ttl time.Duration } // NewRedlock 创建 Redlock func NewRedlock(clients []*redis.Client, ttl time.Duration) *Redlock { return Redlock{ clients: clients, quorum: len(clients)/2 1, ttl: ttl, } } // Lock 获取 Redlock func (r *Redlock) Lock(ctx context.Context) (string, error) { value : generateValue() start : time.Now() successCount : 0 for _, client : range r.clients { // 逐个实例加锁 ok, err : client.SetNX(ctx, value, value, r.ttl).Result() if err nil ok { successCount } } // 计算耗时 elapsed : time.Since(start) // 检查是否满足两个条件 // 1. 成功加锁的实例 法定数量 // 2. 总耗时 TTL防止加锁太久导致锁失效 if successCount r.quorum elapsed r.ttl { log.Printf([Redlock] Acquired: %s (nodes%d/%d, elapsed%v), value, successCount, len(r.clients), elapsed) return value, nil } // 加锁失败释放所有已获取的锁 r.unlockAll(context.Background(), value) return , fmt.Errorf(redlock failed: got %d/%d nodes in %v, successCount, r.quorum, elapsed) } // Unlock 释放 Redlock func (r *Redlock) Unlock(ctx context.Context, value string) error { r.unlockAll(ctx, value) return nil } // unlockAll 在所有实例上释放锁 func (r *Redlock) unlockAll(ctx context.Context, value string) { script : if redis.call(get, KEYS[1]) ARGV[1] then return redis.call(del, KEYS[1]) else return 0 end for _, client : range r.clients { client.Eval(ctx, script, []string{value}, value) } }三、基于 ZooKeeper 的分布式锁ZooKeeper 利用临时顺序节点实现分布式锁// distributed/lock/zklock.go package lock import ( fmt log path strings sync time github.com/go-zookeeper/zk ) // ZKLock ZooKeeper 分布式锁 type ZKLock struct { conn *zk.Conn lockPath string nodePath string // 当前创建的临时顺序节点路径 acl []zk.ACL mu sync.Mutex isLocked bool cancelChan chan struct{} } // NewZKLock 创建 ZooKeeper 锁 func NewZKLock(conn *zk.Conn, lockPath string) *ZKLock { return ZKLock{ conn: conn, lockPath: lockPath, acl: zk.WorldACL(zk.PermAll), cancelChan: make(chan struct{}), } } // Lock 获取锁阻塞直到成功 func (l *ZKLock) Lock() error { l.mu.Lock() defer l.mu.Unlock() // 1. 确保父节点存在 l.ensureParent() // 2. 创建临时顺序节点 nodePath, err : l.conn.Create( path.Join(l.lockPath, lock-), []byte(locked), zk.FlagEphemeral|zk.FlagSequence, l.acl, ) if err ! nil { return fmt.Errorf(create znode: %w, err) } l.nodePath nodePath // 3. 尝试获取锁 if err : l.waitForLock(); err ! nil { return err } l.isLocked true log.Printf([ZKLock] Acquired: %s, nodePath) return nil } // Unlock 释放锁 func (l *ZKLock) Unlock() error { l.mu.Lock() defer l.mu.Unlock() if !l.isLocked { return nil } // 删除临时节点 if err : l.conn.Delete(l.nodePath, -1); err ! nil { return fmt.Errorf(delete znode: %w, err) } l.isLocked false log.Printf([ZKLock] Released: %s, l.nodePath) return nil } // waitForLock 等待获取锁 func (l *ZKLock) waitForLock() error { for { // 获取所有子节点 children, _, err : l.conn.Children(l.lockPath) if err ! nil { return fmt.Errorf(get children: %w, err) } // 排序找出最小的节点 sorted : sortNodes(children) myNode : path.Base(l.nodePath) if sorted[0] myNode { // 我是最小的节点获得锁 return nil } // 监听前一个节点 prevNode : for i, node : range sorted { if node myNode i 0 { prevNode sorted[i-1] break } } if prevNode { return nil } // 监听前一个节点的删除事件 exists, _, events, err : l.conn.ExistsW(path.Join(l.lockPath, prevNode)) if err ! nil { return err } if !exists { // 前一个节点已经消失重新检查 continue } // 阻塞等待事件 select { case event : -events: if event.Type zk.EventNodeDeleted { // 前一个节点被删除重新竞争 continue } case -l.cancelChan: return fmt.Errorf(lock cancelled) } } } // ensureParent 确保父节点存在 func (l *ZKLock) ensureParent() { exists, _, _ : l.conn.Exists(l.lockPath) if !exists { l.conn.Create(l.lockPath, []byte{}, 0, l.acl) } } // sortNodes 对节点名排序提取序号 func sortNodes(nodes []string) []string { // ZooKeeper 的顺序节点格式lock-0000000001 // 按序号升序排列 sorted : make([]string, len(nodes)) copy(sorted, nodes) // 简化实现字符串排序即可序号定长补零 // 实际应提取序号数字排序 return sorted }四、基于 etcd 的分布式锁etcd 利用租约Lease和事务 API 实现分布式锁// distributed/lock/etcd_lock.go package lock import ( context fmt log sync time clientv3 go.etcd.io/etcd/client/v3 go.etcd.io/etcd/client/v3/concurrency ) // EtcdLock etcd 分布式锁使用 concurrency 包 type EtcdLock struct { client *clientv3.Client session *concurrency.Session mutex *concurrency.Mutex key string mu sync.Mutex isLocked bool ttl int // 租约 TTL秒 } // NewEtcdLock 创建 etcd 锁 func NewEtcdLock(client *clientv3.Client, key string, ttl int) *EtcdLock { return EtcdLock{ client: client, key: key, ttl: ttl, } } // Lock 获取锁阻塞直到成功 func (l *EtcdLock) Lock(ctx context.Context) error { l.mu.Lock() defer l.mu.Unlock() // 创建会话自动续期租约 session, err : concurrency.NewSession(l.client, concurrency.WithTTL(l.ttl)) if err ! nil { return fmt.Errorf(create session: %w, err) } l.session session // 创建互斥锁 l.mutex concurrency.NewMutex(l.session, l.key) // 尝试加锁 if err : l.mutex.Lock(ctx); err ! nil { l.session.Close() return fmt.Errorf(etcd lock: %w, err) } l.isLocked true log.Printf([EtcdLock] Acquired: %s (lease%d), l.key, session.Lease()) return nil } // Unlock 释放锁 func (l *EtcdLock) Unlock(ctx context.Context) error { l.mu.Lock() defer l.mu.Unlock() if !l.isLocked { return nil } // 释放锁 if err : l.mutex.Unlock(ctx); err ! nil { return fmt.Errorf(etcd unlock: %w, err) } // 关闭会话释放租约 l.session.Close() l.isLocked false log.Printf([EtcdLock] Released: %s, l.key) return nil } // ─── 手写 etcd 锁不使用 concurrency 包 ─── // RawEtcdLock 手写 etcd 锁 type RawEtcdLock struct { client *clientv3.Client key string value string leaseID clientv3.LeaseID mu sync.Mutex isLocked bool } // Lock 手写 etcd 锁实现 func (l *RawEtcdLock) Lock(ctx context.Context) error { l.mu.Lock() defer l.mu.Unlock() // 1. 创建租约 lease, err : l.client.Grant(ctx, 10) // 10 秒 TTL if err ! nil { return fmt.Errorf(grant lease: %w, err) } l.leaseID lease.ID // 2. 使用事务加锁 // 如果 key 不存在则创建否则失败 txn : l.client.Txn(ctx). If(clientv3.Compare(clientv3.CreateRevision(l.key), , 0)). Then(clientv3.OpPut(l.key, l.value, clientv3.WithLease(lease.ID))). Else() txnResp, err : txn.Commit() if err ! nil { return fmt.Errorf(txn commit: %w, err) } if !txnResp.Succeeded { return fmt.Errorf(lock already held by another client) } // 3. 启动续期协程 go l.keepAlive(ctx) l.isLocked true log.Printf([RawEtcdLock] Acquired: %s, l.key) return nil } // keepAlive 续期租约 func (l *RawEtcdLock) keepAlive(ctx context.Context) { ch, err : l.client.KeepAlive(ctx, l.leaseID) if err ! nil { log.Printf([RawEtcdLock] KeepAlive failed: %v, err) return } for range ch { // 续期成功 } } // Unlock 释放锁 func (l *RawEtcdLock) Unlock(ctx context.Context) error { l.mu.Lock() defer l.mu.Unlock() if !l.isLocked { return nil } // 撤销租约自动删除关联的 key _, err : l.client.Revoke(ctx, l.leaseID) if err ! nil { return fmt.Errorf(revoke lease: %w, err) } l.isLocked false log.Printf([RawEtcdLock] Released: %s, l.key) return nil }五、分布式锁的统一抽象// distributed/lock/locker.go package lock import ( context errors sync time ) // DLocker 分布式锁接口 type DLocker interface { Lock(ctx context.Context) error TryLock(ctx context.Context) (bool, error) Unlock(ctx context.Context) error } // Common errors var ( ErrLockNotHeld errors.New(lock not held by this client) ErrLockTimeout errors.New(lock acquisition timeout) ) // LockOptions 锁配置 type LockOptions struct { Block bool // 是否阻塞等待 Timeout time.Duration // 等待超时 TTL time.Duration // 锁自动过期时间 RetryDelay time.Duration // 重试间隔 } // DefaultLockOptions 默认配置 var DefaultLockOptions LockOptions{ Block: true, Timeout: 5 * time.Second, TTL: 30 * time.Second, RetryDelay: 50 * time.Millisecond, } // LockManager 锁管理器 type LockManager struct { provider DLocker options LockOptions locks map[string]*lockHolder mu sync.Mutex } type lockHolder struct { locker DLocker acquired bool acquireAt time.Time } // NewLockManager 创建锁管理器 func NewLockManager(provider DLocker, opts LockOptions) *LockManager { return LockManager{ provider: provider, options: opts, locks: make(map[string]*lockHolder), } } // Acquire 获取锁 func (lm *LockManager) Acquire(ctx context.Context, key string) error { lm.mu.Lock() if _, exists : lm.locks[key]; exists { lm.mu.Unlock() return fmt.Errorf(lock already held: %s, key) } holder : lockHolder{ locker: lm.provider, acquireAt: time.Now(), } lm.locks[key] holder lm.mu.Unlock() var err error if lm.options.Block { // 阻塞等待 lockCtx, cancel : context.WithTimeout(ctx, lm.options.Timeout) defer cancel() err holder.locker.Lock(lockCtx) } else { // 非阻塞尝试 var ok bool ok, err holder.locker.TryLock(ctx) if err nil !ok { err ErrLockTimeout } } if err ! nil { lm.mu.Lock() delete(lm.locks, key) lm.mu.Unlock() return err } holder.acquired true return nil } // Release 释放锁 func (lm *LockManager) Release(ctx context.Context, key string) error { lm.mu.Lock() holder, exists : lm.locks[key] if !exists { lm.mu.Unlock() return ErrLockNotHeld } delete(lm.locks, key) lm.mu.Unlock() return holder.locker.Unlock(ctx) } // IsHeld 检查是否持有锁 func (lm *LockManager) IsHeld(key string) bool { lm.mu.Lock() defer lm.mu.Unlock() holder, exists : lm.locks[key] return exists holder.acquired }六、三种分布式锁对比特性RedisZooKeeperetcd一致性​AP最终一致CP强一致CP强一致性能​极高纯内存中中高可靠性​低主从切换丢锁高高自动续期​需自己实现临时节点自动租约自动续期羊群效应​无有惊群无Watch 机制运维成本​低高中适用场景​允许偶尔锁失效强一致性要求云原生/K8s 环境七、使用示例// examples/lock_demo.go package main import ( context fmt log sync time distributed/lock ) func main() { fmt.Println( 分布式锁演示 \n) // 模拟秒杀场景10 个并发请求抢 1 件商品 const concurrentRequests 10 var stock int32 1 // 库存 1 件 var soldCount int32 0 var wg sync.WaitGroup fmt.Printf(库存: %d 件\n, stock) fmt.Printf(并发请求: %d 个\n\n, concurrentRequests) // 使用 Redis 锁假设 Redis 已启动 // 实际运行需要 Redis 实例这里展示逻辑 fmt.Println(【场景】Redis 分布式锁保护库存扣减) fmt.Println(伪代码演示实际需要 Redis 连接\n) // 模拟无锁的情况 fmt.Println(--- 无锁会出现超卖 ---) for i : 0; i concurrentRequests; i { wg.Add(1) go func(requestID int) { defer wg.Done() // 没有锁保护 if stock 0 { stock-- fmt.Printf( 请求 %d: 抢购成功\n, requestID) } else { fmt.Printf( 请求 %d: 库存不足\n, requestID) } }(i) } wg.Wait() fmt.Printf( 最终库存: %d (应该为 0 或负数)\n\n, stock) // 恢复库存 stock 1 // 模拟有锁的情况 fmt.Println(--- 有锁安全扣减 ---) var mu sync.Mutex // 模拟分布式锁 for i : 0; i concurrentRequests; i { wg.Add(1) go func(requestID int) { defer wg.Done() // 获取锁模拟 mu.Lock() if stock 0 { stock-- fmt.Printf( 请求 %d: 抢购成功\n, requestID) } else { fmt.Printf( 请求 %d: 库存不足\n, requestID) } mu.Unlock() }(i) } wg.Wait() fmt.Printf( 最终库存: %d (正确应为 0)\n\n, stock) // 三种锁的特性对比 fmt.Println(【三种分布式锁对比】\n) fmt.Println(┌──────────┬──────────────┬──────────────┬──────────────┐) fmt.Println(│ 特性 │ Redis │ ZooKeeper │ etcd │) fmt.Println(├──────────┼──────────────┼──────────────┼──────────────┤) fmt.Println(│ 一致性 │ 最终一致 │ 强一致 │ 强一致 │) fmt.Println(│ 性能 │ 极高 │ 中等 │ 中高 │) fmt.Println(│ 自动续期 │ 需自行实现 │ 临时节点 │ 租约机制 │) fmt.Println(│ 羊群效应 │ 无 │ 有 │ 无 │) fmt.Println(│ 适用场景 │ 允许偶发 │ 强一致 │ 云原生 │) fmt.Println(└──────────┴──────────────┴──────────────┴──────────────┘) fmt.Println(\n演示完成) }八、生产坑位#坑位现象解决方案1锁过期提前释放​业务执行时间超过锁 TTL锁自动释放后被其他线程获取看门狗自动续期 / 合理设置 TTL2Redis 主从切换丢锁​Master 宕机锁还没同步到 Slave新 Master 没有锁记录Redlock 算法 / 改用 etcd3解锁别人的锁​线程 A 释放了线程 B 持有的锁每个锁携带唯一标识解锁时校验4不可重入​同一个线程递归调用时死锁记录线程持有次数重入计数5羊群效应​ZooKeeper 锁释放时唤醒所有等待者只用监听前一个节点公平锁6GC 停顿导致锁超时​Java 世界 Full GC 导致看门狗无法续期设置合理的 TTL / 使用 etcd 租约九、本讲总结┌──────────────────────────────────────────────────────────────────────────┐ │ 第7讲 核心收获 │ │ │ │ ✅ 理解了分布式锁的核心要求 │ │ - 互斥性、安全性防死锁、可重入、容错性 │ │ │ │ ✅ 实现了 Redis 分布式锁 │ │ - SETNX Lua 脚本保证原子性 │ │ - 看门狗自动续期 │ │ - Redlock 解决单点故障 │ │ │ │ ✅ 实现了 ZooKeeper 分布式锁 │ │ - 临时顺序节点实现公平锁 │ │ - Watch 机制监听前一个节点 │ │ │ │ ✅ 实现了 etcd 分布式锁 │ │ - 租约自动续期 │ │ - 事务 API 保证原子性 │ │ │ │ ✅ 对比了三种锁方案的优劣 │ │ ✅ 覆盖了 6 个生产坑位与解决方案 │ │ │ └──────────────────────────────────────────────────────────────────────────┘十、下一讲预告第8讲《分布式链路追踪》你将学到OpenTelemetry 标准Trace/Span/Context 传播Jaeger/Zipkin 集成采样策略头采样/尾采样从零实现一个链路追踪 SDK 开发之余的小工具推荐在调试分布式锁的超时和续期问题时zz365.top 的时间计算器可以精确计算锁 TTL 与业务执行时间的差值帮助你确定合理的 TTL 阈值。其 UUID 生成器则可以为每个锁生成唯一的标识值确保解锁时的安全性。所有处理均在浏览器本地完成不会泄露你的业务数据。