Go和Redis实现分布式锁

本文介绍分布式锁的问题，以及分布式锁实现的方法。

为什么要使用锁，问题的引入？

在 进程、线程、协程介绍 一文中我们介绍了进程和线程，从文章中能了解到线程共享进程的内存全局变量，那么对于全局变量数据一致性的要求，需要在进程内对修改行为加锁以创造临界区。在分布式系统中，发并操作会导致数据不一致的情况，接下要解决的几个问题：

• 单个应用中使用锁：（单进程多线程）
• 分布式锁控制分布式系统之间同步访问资源的一种方式（相同的应用部署多套，并发请求，会执行到相同代码块）。

设计一个分布式锁需要具备哪些条件？

• 互斥性：在任意一个时刻，只有一个客户端持有锁。
• 无死锁：即便持有锁的客户端崩溃或者其他意外事件，锁仍然可以被获取。
• 容错：以redis为例子，只要大部分Redis节点都活着，客户端就可以获取和释放锁。

在单机程序并发操作中我们做一下锁的实验，如下对全局变量的自增操作。

不加锁代码演示：

结果：多次运行会得到不同的结果；

package main 
import ( 
    "sync" 
) 
// 全局变量 
var counter int 
func main() { 
    var wg sync.WaitGroup 
    for i := 0; i < 1000; i++ { 
        wg.Add(1) 
        go func() { 
        defer wg.Done() 
            counter++ 
        }() 
    } 
    wg.Wait() 
    println(counter) 
}
多次运行会得到不同的结果：
go run local_lock.go 
945 
go run local_lock.go 
937 
go run local_lock.go 
959

加锁代码的演示：

结果：数据保持一致；

package main 
import ( 
    "sync" 
) 
// 全局变量 
var counter int 
var l sync.Mutex
func main() { 
    var wg sync.WaitGroup 
    for i := 0; i < 1000; i++ { 
        wg.Add(1) 
        go func() { 
        	  defer wg.Done()
           l.Lock()
            counter++ 
           l.Unlock()
        }() 
    } 
    wg.Wait() 
    println(counter) 
}
多次运行会得到不同的结果：
go run local_lock.go 
1000

在某些场景，我们只是希望一个任务有单一的执行者，而不像计数器场景那样，所有Goroutine都执行成功。后来的Goroutine在抢锁失败后，需要放弃其流程。这时候就需要尝试锁（try lock）了。

顾名思义，尝试锁如果加锁成功执行后续流程，如果加锁失败也不会阻塞，而会直接返回加锁的结果。在Go语言中可以用大小为1的通道模拟尝试锁：

解决问题 ：一个任务有单一的执行者

结果：如果加锁失败也不会阻塞，而会直接返回加锁的结果；

package main 
import ( 
    "sync" 
) 
// Lock 尝试锁 
type Lock struct { 
    c chan struct{} 
} 
// NewLock 生成一个尝试锁 
func NewLock() Lock { 
    var l Lock 
    l.c = make(chan struct{}, 1) 
    l.c <- struct{}{} 
    return l 
} 
// Lock 锁住尝试锁返回加锁结果 
func (l Lock) Lock() bool { 
    lockResult := false 
    select { 
    case <-l.c: 
        lockResult = true 
    default: 
    } 
    return lockResult 
} 
// Unlock 解锁尝试锁 
func (l Lock) Unlock() { 
    l.c <- struct{}{} 
} 
var counter int 
func main() { 
    var l = NewLock() 
    var wg sync.WaitGroup 
    for i := 0; i < 10; i++ { 
        wg.Add(1) 
        go func() { 
            defer wg.Done() 
            if !l.Lock() { 
                // log error 
                println("lock failed") 
                return 
            } 
            counter++ 
            println("current counter", counter) 
            l.Unlock() 
        }() 
    } 
    wg.Wait() 
}

因为我们的逻辑限定每个Goroutine只有成功执行了Lock才会继续执行后续逻辑，因此在Unlock时可以保证Lock结构体中的通道一定是空，从而不会阻塞，也不会失败。上面的代码使用了大小为1的通道来模拟尝试锁，理论上还可以使用标准库中的CAS来实现相同的功能且成本更低，读者可以自行尝试。

在单机系统中，尝试锁并不是一个好选择，因为大量的Goroutine抢锁可能会导致CPU无意义的资源浪费。有一个专有名词用来描述这种抢锁的场景——活锁。

活锁指的是程序看起来在正常执行，但实际上CPU周期被浪费在抢锁而非执行任务上，从而导致程序整体的执行效率低下。活锁的问题定位起来要麻烦很多，所以在单机场景下，不建议使用这种锁。

解决问题：分布式锁控制分布式系统之间同步访问资源的一种方式。

基于 Redis 的 SETNX 指令完成锁的获取。

获取锁

SetNX：当一个线程执行 setnx 返回 1，说明 key 原本不存在，该线程成功得到了锁；当一个线程执行 setnx 返回 0，说明 key 已经存在，该线程抢锁失败。

释放锁

Del：检查解锁方是否是加锁方，是则允许解锁，否则不允许解锁。

package main 
import ( 
    "fmt" 
    "sync" 
    "time" 
    "github.com/go-redis/redis" 
) 
func incr() { 
    client := redis.NewClient(&redis.Options{ 
        Addr:     "localhost:6379", 
        Password: "", // no password set 
        DB:       0,  // use default DB 
    }) 
    var lockKey = "counter_lock" 
    var counterKey = "counter" 
    // lock 
    resp := client.SetNX(lockKey, 1, time.Second*5) 
    lockSuccess, err := resp.Result() 
    if err != nil || !lockSuccess { 
        fmt.Println(err, "lock result: ", lockSuccess) 
        return 
    } 
    // counter ++ 
    getResp := client.Get(counterKey) 
    cntValue, err := getResp.Int64() 
    if err == nil { 
        cntValue++ 
        resp := client.Set(counterKey, cntValue, 0) 
        _, err := resp.Result() 
        if err != nil { 
            // log err 
            println("set value error!") 
        } 
    } 
    println("current counter is ", cntValue) 
    delResp := client.Del(lockKey) 
    unlockSuccess, err := delResp.Result() 
    if err == nil && unlockSuccess > 0 { 
        println("unlock success!") 
    } else { 
        println("unlock failed", err) 
    } 
} 
func main() { 
    var wg sync.WaitGroup 
    for i := 0; i < 10; i++ { 
        wg.Add(1) 
        go func() { 
            defer wg.Done() 
            incr() 
        }() 
    } 
    wg.Wait() 
}
看看运行结果：

通过代码和执行结果可以看到，远程调用setnx实际上和单机的尝试锁非常相似，如果获取锁失败，那么相关的任务逻辑就不应该继续向前执行。

setnx很适合在高并发场景下，用来争抢一些“唯一”的资源。例如，交易撮合系统中卖家发起订单，而多个买家会对其进行并发争抢。这种场景我们没有办法依赖具体的时间来判断先后，因为不管是用户设备的时间，还是分布式场景下的各台机器的时间，都是没有办法在合并后保证正确的时序的。哪怕是同一个机房的集群，不同的机器的系统时间可能也会有细微的差别。

所以，我们需要依赖这些请求到达Redis节点的顺序来做正确的抢锁操作。如果用户的网络环境比较差，那也只能自求多福了。

通过上面的方式，我们好像是解决了分布式锁的问题，但是想想还有没有什么问题呢？？

对，问题还是有的，可能会有死锁的问题发生，比如服务器1设置完之后，获取了锁之后，忽然发生了宕机。

那后续的删除key操作就没法执行，这个key会一直在redis中存在，其他服务器每次去检查，都会返回0，他们都会认为有人在使用锁，我需要等。

为了解决这个死锁的问题，我们就就需要给key 设置有效期了。

第一种就是在set完key之后，直接设置key的有效期 "expire key timeout" ，为key设置一个超时时间，单位为second，超过这个时间锁会自动释放，避免死锁。

这种方式相当于，把锁持有的有效期，交给了redis去控制。如果时间到了，你还没有给我删除key，那redis就直接给你删了，其他服务器就可以继续去setnx获取锁。

第二种方式，就是把删除key权利交给其他的服务器，那这个时候就需要用到value值了

比如服务器1，设置了value 也就是 timeout 为 当前时间+1 秒 ，这个时候服务器2 通过get 发现时间已经超过系统当前时间了，那就说明服务器1没有释放锁，服务器1可能出问题了，

服务器2就开始执行删除key操作，并且继续执行setnx 操作。

但是这块有一个问题，也就是，不光你服务器2可能会发现服务器1超时了，服务器3也可能会发现，如果刚好，服务器2，setnx操作完成，服务器3就接着删除，是不是服务器3也可以setnx成功了？

那就等于是服务器2和服务器3都拿到锁了，那就问题大了。这个时候怎么办呢？

这个时候需要用到 “GETSET key value” 命令了。这个命令的意思就是获取当前key的值，并且设置新的值。

假设服务器2发现key过期了，开始调用 getset 命令，然后用获取的时间判断是否过期，如果获取的时间仍然是过期的，那就说明拿到锁了。

如果没有，则说明在服务2执行getset之前，服务器3可能也发现锁过期了，并且在服务器2之前执行了getset操作，重新设置了过期时间。

那么服务器2就需要放弃后续的操作，继续等待服务器3释放锁或者去监测key的有效期是否过期。

这块其实有一个小问题是，服务器3已经修改了有效期，拿到锁之后，服务器2，也修改了有效期，但是没能拿到锁，但是这个有效期的时间已经被在服务器3的基础上有增加一些，但是这种影响其实还是很小的，几乎可以忽略不计。