本文介绍分布式锁的问题，以及分布式锁实现的方法。
为什么要使用锁，问题的引入？
在 一文中我们介绍了进程和线程，从文章中能了解到线程共享进程的 内存全局变量 ，那么对于全局变量数据一致性的要求，需要在进程内对修改行为加锁以创造临界区。在分布式系统中，发并操作会导致数据不一致的情况，接下要解决的几个问题：
- 单个应用中使用锁：（单进程多线程）
- 分布式锁控制分布式系统之间同步访问资源的一种方式（相同的应用部署多套，并发请求，会执行到相同代码块）。
设计一个分布式锁需要具备哪些条件？
- 互斥性：在任意一个时刻，只有一个客户端持有锁。
- 无死锁：即便持有锁的客户端崩溃或者其他意外事件，锁仍然可以被获取。
- 容错：以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的基础上有增加一些，但是这种影响其实还是很小的，几乎可以忽略不计。
