Go微服务: redis分布式锁保证数据原子操作的一致性

概述

随着云计算和大数据技术的飞速发展，分布式系统已经成为现代IT架构的重要组成部分
在分布式系统中，数据的一致性是一个至关重要的挑战，特别是在并发访问和修改共享资源的场景下
分布式锁是一种跨进程、跨机器节点的互斥锁，用于控制多个节点对共享资源的访问
其核心目标是确保在分布式系统中，同一时刻只有一个节点能够访问特定的共享资源，从而实现数据的一致性
分布式锁的实现方式多种多样，包括基于数据库、缓存（如Redis）、分布式协调服务（如Zookeeper）等

场景示例

在多携程或者多线程的情况下，这个数据竞争是避免不了的, 参考下图

一开始我们有一个Redis图标代表着之前用到的分布式的锁
两边有两个grorouting，在实际工作当中有很多grorouting
让左边的 grorouting 进行 getKey 和 setKey，就是说对这个 redis 进行读写
让右边的 grorouting 也进行读写，这里面就会产生一个问题
多个gorouting在同时写的时候会不会产生数据一致性的问题
这个场景就是我们经常说的中断，举个例子
- 写代码，听歌上网写文档，感觉这个计算机同时在做几件事情一样
- 但是要在计算机看来，它在不停的切换，只是说计算机执行的足够快
- 人类是无法感觉到它在极短的时间内进行时间的这个切换
这里有两个命令，一个是上面的 getKey和下面这一行的 setKey
在高并发的时候就会导致切换出现数据竞争, 也就是数据不一致的这种情况发生
REDIS 里提供了一个命令就是这个 Setnx，全称就是 Set if Not Exists
中文意思就是说设置它，如果它不存在，如果设置成功，就是1，设置失败就返回0
那就是说不存在这个key的时候，我就可以设置成功，如果存在了，那就设置是失败的
那这就保证了第一个gorouting是可以设置成功的，在这个后面的gorouting，它就是设置不成功的

源码示例

分布式锁 redSync 是如何结合这个setNX解决这个数据竞争的问题的

我们来看下之前的业务代码

pool := goredis.NewPool(client)
rs := redsync.New(pool)
mutexname := "my-global-mutex"
mutex := rs.NewMutex(mutexname, redsync.WithExpiry(30*time.Second))
fmt.Println("Lock()....")
if err := mutex.Lock(); err != nil {panic(err)
}

上面这里 mutex.Lock() 进入源码

// Lock locks m. In case it returns an error on failure, you may retry to acquire the lock by calling this method again.
func (m *Mutex) Lock() error {return m.LockContext(context.Background())
}

这里看到就一个 LockContext 方法，再次进入，其实这里有2个重载函数

// LockContext locks m. In case it returns an error on failure, you may retry to acquire the lock by calling this method again.
func (m *Mutex) LockContext(ctx context.Context) error {return m.lockContext(ctx, m.tries)
}// lockContext locks m. In case it returns an error on failure, you may retry to acquire the lock by calling this method again.
func (m *Mutex) lockContext(ctx context.Context, tries int) error {// 如果 ctx 不存在则新建if ctx == nil {ctx = context.Background()}// 获取 valuevalue, err := m.genValueFunc()if err != nil {return err}var timer *time.Timer// 循环 tries 次for i := 0; i < tries; i++ {if i != 0 {if timer == nil {timer = time.NewTimer(m.delayFunc(i))} else {timer.Reset(m.delayFunc(i))}select {case <-ctx.Done():timer.Stop()// Exit early if the context is done.return ErrFailedcase <-timer.C:// Fall-through when the delay timer completes.}}start := time.Now()n, err := func() (int, error) {ctx, cancel := context.WithTimeout(ctx, time.Duration(int64(float64(m.expiry)*m.timeoutFactor)))defer cancel()return m.actOnPoolsAsync(func(pool redis.Pool) (bool, error) {// 在分配锁的时候会加 nx, 进入这里return m.acquire(ctx, pool, value)})}()now := time.Now()until := now.Add(m.expiry - now.Sub(start) - time.Duration(int64(float64(m.expiry)*m.driftFactor)))if n >= m.quorum && now.Before(until) {m.value = valuem.until = untilreturn nil}_, _ = func() (int, error) {ctx, cancel := context.WithTimeout(ctx, time.Duration(int64(float64(m.expiry)*m.timeoutFactor)))defer cancel()return m.actOnPoolsAsync(func(pool redis.Pool) (bool, error) {return m.release(ctx, pool, value)})}()if i == tries-1 && err != nil {return err}}return ErrFailed
}

再次进入 m.acquire

func (m *Mutex) acquire(ctx context.Context, pool redis.Pool, value string) (bool, error) {conn, err := pool.Get(ctx)if err != nil {return false, err}defer conn.Close()// 这里 SetNX 就是 redSync 封装起来的 nx 特性reply, err := conn.SetNX(m.name, value, m.expiry)if err != nil {return false, err}return reply, nil
}

进入 conn.SetNX，可见 key, value 和过期时间 expiry (默认 8s)
```
// Conn is a single Redis connection.
type Conn interface {Get(name string) (string, error)Set(name string, value string) (bool, error)SetNX(name string, value string, expiry time.Duration) (bool, error) // 注意这里Eval(script *Script, keysAndArgs ...interface{}) (interface{}, error)PTTL(name string) (time.Duration, error)Close() error
}
```
- 不传过期时间，它会有死锁的风险
- 因为一旦处于某种异常状况，那你这个锁就永远不会释放了，就会出现死锁
- 过期时间比如说五秒之后自动释放，那我无论是宕机还是崩溃等其他问题，它都会五秒之后，释放出资源
- 释放之后其他 gorouting 就可以继续抢锁和业务逻辑操作

关于redis锁过期问题

在上述业务代码中

mutex := rs.NewMutex(mutexname, redsync.WithExpiry(30*time.Second))

这里有对过期时间做初始化的代码，其实在new的时候，有一个默认值，进入 NewMutex

// NewMutex returns a new distributed mutex with given name.
func (r *Redsync) NewMutex(name string, options ...Option) *Mutex {m := &Mutex{name:   name,expiry: 8 * time.Second,tries:  32,delayFunc: func(tries int) time.Duration {return time.Duration(rand.Intn(maxRetryDelayMilliSec-minRetryDelayMilliSec)+minRetryDelayMilliSec) * time.Millisecond},genValueFunc:  genValue,driftFactor:   0.01,timeoutFactor: 0.05,quorum:        len(r.pools)/2 + 1,pools:         r.pools,}for _, o := range options {o.Apply(m)}if m.shuffle {randomPools(m.pools)}return m
}

这里可见，默认是8s的过期时间，如果业务特别复杂，这个时间可以调整，因为如果8s的时间hold不住业务，其他gorouting就可能抢到锁干一些事情导致数据不一致

redSync 中有一种机制是对锁进行续命，在 mutux.go 中有个 touch 方法

// 里面不是go脚本，是 Lua 脚本
var touchScript = redis.NewScript(1, `// 核对你是否是key和值的拥有者，不能让别人随便设置// 只有自己才能对自己进行续期if redis.call("GET", KEYS[1]) == ARGV[1] thenreturn redis.call("PEXPIRE", KEYS[1], ARGV[2])elsereturn 0end
`)func (m *Mutex) touch(ctx context.Context, pool redis.Pool, value string, expiry int) (bool, error) {conn, err := pool.Get(ctx)if err != nil {return false, err}defer conn.Close()touchScript := touchScriptif m.setNXOnExtend {touchScript = touchWithSetNXScript}status, err := conn.Eval(touchScript, m.name, value, expiry)if err != nil {return false, err}return status != int64(0), nil
}

也就是说默认8s后，如果不够，通过 touch 方法续期，如果一直续期则会导致锁会一直被占用，结果带来的是不公平和可能得死锁问题需要谨慎
一般在8s内已经足够业务使用了，如果不够，可以考虑使用，但一定要慎用

分布式锁如何防止被篡改和删除

redSync 已经对防篡改和删除做了一些防范处理

我们可以从 Unlock 的源码中获得一些答案

// Unlock unlocks m and returns the status of unlock.
func (m *Mutex) Unlock() (bool, error) {return m.UnlockContext(context.Background())
}// UnlockContext unlocks m and returns the status of unlock.
func (m *Mutex) UnlockContext(ctx context.Context) (bool, error) {n, err := m.actOnPoolsAsync(func(pool redis.Pool) (bool, error) {return m.release(ctx, pool, m.value)})if n < m.quorum {return false, err}return true, nil
}

进入 release 函数

var deleteScript = redis.NewScript(1, `local val = redis.call("GET", KEYS[1])if val == ARGV[1] thenreturn redis.call("DEL", KEYS[1])elseif val == false thenreturn -1elsereturn 0end
`)func (m *Mutex) release(ctx context.Context, pool redis.Pool, value string) (bool, error) {conn, err := pool.Get(ctx)if err != nil {return false, err}defer conn.Close()status, err := conn.Eval(deleteScript, m.name, value)if err != nil {return false, err}if status == int64(-1) {return false, ErrLockAlreadyExpired}return status != int64(0), nil
}

这里的核心是 conn.Eval(deleteScript, m.name, value) 直接把值删除，在删除的逻辑又是一段 Lua
Lua可以保证原子性，性能更高，现在再看这个 value 是怎么来的，回到 NewMutex 中，value 就在这里拿到的，在其源码内部有一个 genValueFunc: genValue 的配置项
```
func genValue() (string, error) {b := make([]byte, 16)_, err := rand.Read(b)if err != nil {return "", err}return base64.StdEncoding.EncodeToString(b), nil
}
```
这里有一个处理 base64 的程序，其实在内部每次调用一次 LockContext 都会执行这个函数生成一次这个 base64并存放到 m.value 上，这个base64 就达到了防篡改的目的
回到 Unlock 删除时的 deleteScript 中可以看到，删除时也要保证是自己的，这样就保证了数据的安全性
这样就不会随随便便被别人给干掉