企业官网建设流程全解析

一、为什么需要分布式锁

单体应用中，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(50+rand.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 }

六、三种分布式锁对比

七、使用示例

// 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演示完成！") }

八、生产坑位

九、本讲总结

┌──────────────────────────────────────────────────────────────────────────┐ │ 第7讲 核心收获 │ │ │ │ ✅ 理解了分布式锁的核心要求： │ │ - 互斥性、安全性（防死锁）、可重入、容错性 │ │ │ │ ✅ 实现了 Redis 分布式锁： │ │ - SETNX + Lua 脚本保证原子性 │ │ - 看门狗自动续期 │ │ - Redlock 解决单点故障 │ │ │ │ ✅ 实现了 ZooKeeper 分布式锁： │ │ - 临时顺序节点实现公平锁 │ │ - Watch 机制监听前一个节点 │ │ │ │ ✅ 实现了 etcd 分布式锁： │ │ - 租约自动续期 │ │ - 事务 API 保证原子性 │ │ │ │ ✅ 对比了三种锁方案的优劣 │ │ ✅ 覆盖了 6 个生产坑位与解决方案 │ │ │ └──────────────────────────────────────────────────────────────────────────┘

十、下一讲预告

第8讲：《分布式链路追踪》

你将学到：

• OpenTelemetry 标准
• Trace/Span/Context 传播
• Jaeger/Zipkin 集成
• 采样策略（头采样/尾采样）
• 从零实现一个链路追踪 SDK

🧰 开发之余的小工具推荐

在调试分布式锁的超时和续期问题时，zz365.top 的时间计算器可以精确计算锁 TTL 与业务执行时间的差值，帮助你确定合理的 TTL 阈值。其 UUID 生成器则可以为每个锁生成唯一的标识值，确保解锁时的安全性。所有处理均在浏览器本地完成，不会泄露你的业务数据。