40分钟学 Go 语言高并发：分布式锁实现

分布式锁实现

一、概述

分布式锁是分布式系统中的一个重要组件，用于协调分布式环境下的资源访问和并发控制。我们将从锁设计、死锁预防、性能优化和容错处理四个维度深入学习。

学习目标

维度	重点内容	掌握程度
锁设计	基于Redis/etcd的锁实现原理	必须掌握
死锁预防	超时机制、重入机制	必须掌握
性能优化	锁粒度控制、读写分离	重点掌握
容错处理	节点故障、网络分区	重点掌握

二、实现流程图

在这里插入图片描述

三、基础锁实现

让我们首先实现一个基于Redis的分布式锁基础版本：

package distlockimport ("context""crypto/rand""encoding/base64""errors""time""github.com/go-redis/redis/v8"
)type DistributedLock struct {client     *redis.Clientkey        stringvalue      stringexpiration time.Duration
}// NewDistributedLock 创建一个新的分布式锁实例
func NewDistributedLock(client *redis.Client, key string, expiration time.Duration) (*DistributedLock, error) {// 生成随机值作为锁的标识b := make([]byte, 16)_, err := rand.Read(b)if err != nil {return nil, err}value := base64.StdEncoding.EncodeToString(b)return &DistributedLock{client:     client,key:        key,value:      value,expiration: expiration,}, nil
}// TryLock 尝试获取锁
func (dl *DistributedLock) TryLock(ctx context.Context) (bool, error) {return dl.client.SetNX(ctx, dl.key, dl.value, dl.expiration).Result()
}// Unlock 释放锁
func (dl *DistributedLock) Unlock(ctx context.Context) error {script := `if redis.call("get", KEYS[1]) == ARGV[1] thenreturn redis.call("del", KEYS[1])elsereturn 0end`result, err := dl.client.Eval(ctx, script, []string{dl.key}, dl.value).Result()if err != nil {return err}if result == 0 {return errors.New("lock not held")}return nil
}// RefreshLock 刷新锁的过期时间
func (dl *DistributedLock) RefreshLock(ctx context.Context) error {script := `if redis.call("get", KEYS[1]) == ARGV[1] thenreturn redis.call("pexpire", KEYS[1], ARGV[2])elsereturn 0end`result, err := dl.client.Eval(ctx,script,[]string{dl.key},dl.value,dl.expiration.Milliseconds(),).Result()if err != nil {return err}if result == 0 {return errors.New("lock not held")}return nil
}// IsLocked 检查锁是否被持有
func (dl *DistributedLock) IsLocked(ctx context.Context) (bool, error) {exists, err := dl.client.Exists(ctx, dl.key).Result()if err != nil {return false, err}return exists == 1, nil
}

四、增强版锁实现（带可重入特性）

下面是一个支持可重入的分布式锁实现：

package distlockimport ("context""encoding/json""errors""sync""time""github.com/go-redis/redis/v8"
)type LockInfo struct {Owner     string `json:"owner"`Count     int    `json:"count"`Timestamp int64  `json:"timestamp"`
}type ReentrantLock struct {client     *redis.Clientkey        stringowner      stringexpiration time.Durationmu         sync.Mutex
}// NewReentrantLock 创建可重入锁
func NewReentrantLock(client *redis.Client, key string, owner string, expiration time.Duration) *ReentrantLock {return &ReentrantLock{client:     client,key:        key,owner:      owner,expiration: expiration,}
}// Lock 获取可重入锁
func (rl *ReentrantLock) Lock(ctx context.Context) error {rl.mu.Lock()defer rl.mu.Unlock()script := `local lockInfo = redis.call('get', KEYS[1])if not lockInfo then-- 锁不存在，创建新锁redis.call('set', KEYS[1], ARGV[1], 'PX', ARGV[2])return 1endlocal info = cjson.decode(lockInfo)if info.owner == ARGV[3] then-- 重入锁info.count = info.count + 1info.timestamp = tonumber(ARGV[4])redis.call('set', KEYS[1], cjson.encode(info), 'PX', ARGV[2])return 1endreturn 0`lockInfo := LockInfo{Owner:     rl.owner,Count:     1,Timestamp: time.Now().UnixNano(),}lockInfoJSON, err := json.Marshal(lockInfo)if err != nil {return err}result, err := rl.client.Eval(ctx,script,[]string{rl.key},string(lockInfoJSON),rl.expiration.Milliseconds(),rl.owner,time.Now().UnixNano(),).Result()if err != nil {return err}if result.(int64) == 0 {return errors.New("failed to acquire lock")}return nil
}// Unlock 释放可重入锁
func (rl *ReentrantLock) Unlock(ctx context.Context) error {rl.mu.Lock()defer rl.mu.Unlock()script := `local lockInfo = redis.call('get', KEYS[1])if not lockInfo thenreturn 0endlocal info = cjson.decode(lockInfo)if info.owner ~= ARGV[1] thenreturn -1endinfo.count = info.count - 1if info.count <= 0 thenredis.call('del', KEYS[1])return 1elseredis.call('set', KEYS[1], cjson.encode(info), 'PX', ARGV[2])return 1end`result, err := rl.client.Eval(ctx,script,[]string{rl.key},rl.owner,rl.expiration.Milliseconds(),).Result()if err != nil {return err}switch result.(int64) {case -1:return errors.New("lock held by another owner")case 0:return errors.New("lock not held")default:return nil}
}// RefreshLock 刷新锁的过期时间
func (rl *ReentrantLock) RefreshLock(ctx context.Context) error {script := `local lockInfo = redis.call('get', KEYS[1])if not lockInfo thenreturn 0endlocal info = cjson.decode(lockInfo)if info.owner ~= ARGV[1] thenreturn 0endinfo.timestamp = tonumber(ARGV[3])redis.call('set', KEYS[1], cjson.encode(info), 'PX', ARGV[2])return 1`result, err := rl.client.Eval(ctx,script,[]string{rl.key},rl.owner,rl.expiration.Milliseconds(),time.Now().UnixNano(),).Result()if err != nil {return err}if result.(int64) == 0 {return errors.New("lock not held")}return nil
}

五、死锁预防机制

1. 超时机制

所有锁操作都设置了过期时间
使用看门狗机制自动续期
防止客户端崩溃导致的死锁

2. 死锁检测

检测项	处理方式	实现难度
循环等待	资源有序分配	中等
持有等待	一次性申请所有资源	简单
不可剥夺	超时自动释放	简单
互斥访问	读写分离	较难

六、性能优化策略

1. 锁粒度优化

降低锁粒度，提高并发度
使用多粒度锁机制
实现分段锁

2. 读写分离

package distlockimport ("context""fmt""time""github.com/go-redis/redis/v8"
)type RWLock struct {client     *redis.Clientkey        stringowner      stringexpiration time.Duration
}func NewRWLock(client *redis.Client, key string, owner string, expiration time.Duration) *RWLock {return &RWLock{client:     client,key:        key,owner:      owner,expiration: expiration,}
}// RLock 获取读锁
func (rwl *RWLock) RLock(ctx context.Context) error {script := `-- 检查是否存在写锁if redis.call('exists', KEYS[1] .. ':write') == 1 thenreturn 0end-- 增加读锁计数local count = redis.call('incr', KEYS[1] .. ':read')redis.call('pexpire', KEYS[1] .. ':read', ARGV[1])-- 记录读锁持有者redis.call('hset', KEYS[1] .. ':readers', ARGV[2], '1')redis.call('pexpire', KEYS[1] .. ':readers', ARGV[1])return 1`result, err := rwl.client.Eval(ctx,script,[]string{rwl.key},rwl.expiration.Milliseconds(),rwl.owner,).Result()if err != nil {return fmt.Errorf("failed to acquire read lock: %v", err)}if result.(int64) == 0 {return fmt.Errorf("write lock exists")}return nil
}// RUnlock 释放读锁
func (rwl *RWLock) RUnlock(ctx context.Context) error {script := `-- 检查读锁是否存在if redis.call('exists', KEYS[1] .. ':read') == 0 thenreturn 0end-- 检查当前客户端是否持有读锁if redis.call('hexists', KEYS[1] .. ':readers', ARGV[1]) == 0 thenreturn -1end-- 移除读锁持有者记录redis.call('hdel', KEYS[1] .. ':readers', ARGV[1])-- 减少读锁计数local count = redis.call('decr', KEYS[1] .. ':read')if count <= 0 thenredis.call('del', KEYS[1] .. ':read')redis.call('del', KEYS[1] .. ':readers')endreturn 1`result, err := rwl.client.Eval(ctx,script,[]string{rwl.key},rwl.owner,).Result()if err != nil {return fmt.Errorf("failed to release read lock: %v", err)}switch result.(int64) {case -1:return fmt.Errorf("read lock not held by this client")case 0:return fmt.Errorf("read lock not exists")default:return nil}
}// Lock 获取写锁
func (rwl *RWLock) Lock(ctx context.Context) error {script := `-- 检查是否存在读锁或写锁if redis.call('exists', KEYS[1] .. ':read') == 1 orredis.call('exists', KEYS[1] .. ':write') == 1 thenreturn 0end-- 设置写锁redis.call('set', KEYS[1] .. ':write', ARGV[1], 'PX', ARGV[2])return 1`result, err := rwl.client.Eval(ctx,script,[]string{rwl.key},rwl.owner,rwl.expiration.Milliseconds(),).Result()if err != nil {return fmt.Errorf("failed to acquire write lock: %v", err)}if result.(int64) == 0 {return fmt.Errorf("lock exists")}return nil
}// Unlock 释放写锁
func (rwl *RWLock) Unlock(ctx context.Context) error {script := `-- 检查写锁是否存在且属于当前客户端local value = redis.call('get', KEYS[1] .. ':write')if not value thenreturn 0endif value ~= ARGV[1] thenreturn -1end-- 删除写锁redis.call('del', KEYS[1] .. ':write')return 1`result, err := rwl.client.Eval(ctx,script,[]string{rwl.key},rwl.owner,).Result()if err != nil {return fmt.Errorf("failed to release write lock: %v", err)}switch result.(int64) {case -1:return fmt.Errorf("write lock not held by this client")case 0:return fmt.Errorf("write lock not exists")default:return nil}
}

七、容错处理

1. 容错机制设计

在这里插入图片描述

2. 故障处理实现

package distlockimport ("context""errors""sync""time""github.com/go-redis/redis/v8"
)type FaultTolerantLock struct {master     *redis.Clientslaves     []*redis.ClientlocalLock  sync.Mutexkey        stringowner      stringexpiration time.Duration
}func NewFaultTolerantLock(master *redis.Client,slaves []*redis.Client,key string,owner string,expiration time.Duration,
) *FaultTolerantLock {return &FaultTolerantLock{master:     master,slaves:     slaves,key:        key,owner:      owner,expiration: expiration,}
}// Lock 获取容错锁
func (ftl *FaultTolerantLock) Lock(ctx context.Context) error {// 1. 尝试在主节点获取锁if err := ftl.tryLockOnMaster(ctx); err == nil {return nil}// 2. 主节点失败，尝试在从节点获取锁if err := ftl.tryLockOnSlaves(ctx); err == nil {return nil}// 3. 所有Redis节点都失败，降级使用本地锁ftl.localLock.Lock()// 4. 启动后台协程尝试恢复到Redis锁go ftl.tryRecoverToRedis(context.Background())return nil
}func (ftl *FaultTolerantLock) tryLockOnMaster(ctx context.Context) error {script := `if redis.call('exists', KEYS[1]) == 0 thenredis.call('set', KEYS[1], ARGV[1], 'PX', ARGV[2])return 1endreturn 0`result, err := ftl.master.Eval(ctx,script,[]string{ftl.key},ftl.owner,ftl.expiration.Milliseconds(),).Result()if err != nil {return err}if result.(int64) == 0 {return errors.New("lock exists")}return nil
}func (ftl *FaultTolerantLock) tryLockOnSlaves(ctx context.Context) error {// 需要在多数从节点上获取锁才算成功successCount := 0majorityCount := (len(ftl.slaves) / 2) + 1for _, slave := range ftl.slaves {if err := ftl.tryLockOnNode(ctx, slave); err == nil {successCount++if successCount >= majorityCount {return nil}}}return errors.New("failed to acquire lock on majority of slaves")
}func (ftl *FaultTolerantLock) tryLockOnNode(ctx context.Context, node *redis.Client) error {script := `if redis.call('exists', KEYS[1]) == 0 thenredis.call('set', KEYS[1], ARGV[1], 'PX', ARGV[2])return 1endreturn 0`result, err := node.Eval(ctx,script,[]string{ftl.key},ftl.owner,ftl.expiration.Milliseconds(),).Result()if err != nil {return err}if result.(int64) == 0 {return errors.New("lock exists")}return nil
}func (ftl *FaultTolerantLock) tryRecoverToRedis(ctx context.Context) {ticker := time.NewTicker(time.Second)defer ticker.Stop()for {select {case <-ctx.Done():returncase <-ticker.C:// 尝试恢复到Redis主节点if err := ftl.tryLockOnMaster(ctx); err == nil {ftl.localLock.Unlock()return}// 尝试恢复到Redis从节点if err := ftl.tryLockOnSlaves(ctx); err == nil {ftl.localLock.Unlock()return}}}
}// Unlock 释放锁
func (ftl *FaultTolerantLock) Unlock(ctx context.Context) error {// 尝试释放Redis锁if err := ftl.unlockRedis(ctx); err == nil {return nil}// Redis释放失败，释放本地锁ftl.localLock.Unlock()return nil
}func (ftl *FaultTolerantLock) unlockRedis(ctx context.Context) error {script := `if redis.call('get', KEYS[1]) == ARGV[1] thenreturn redis.call('del', KEYS[1])endreturn 0`// 先尝试在主节点释放result, err := ftl.master.Eval(ctx,script,[]string{ftl.key},ftl.owner,).Result()if err == nil && result.(int64) == 1 {return nil}// 主节点释放失败，尝试在从节点释放for _, slave := range ftl.slaves {result, err = slave.Eval(ctx,script,[]string{ftl.key},ftl.owner,).Result()if err == nil && result.(int64) == 1 {return nil}}return errors.New("failed to release lock on all nodes")
}

八、性能测试与监控

1. 性能指标

指标	说明	目标值
获取锁延迟	从发起请求到获取锁的时间	<50ms
释放锁延迟	从发起释放到完成的时间	<30ms
锁冲突率	获取锁失败的比例	<10%
QPS	每秒处理的锁请求数	>1000