go分布式锁的一个简单实现

其实锁这种东西，都能能不加就不加，锁会导致程序一定程度上退回到串行化，进而降低效率。


首先，看一个案例，如果要实现一个计数器，并且是多个协程共同进行的，就会出现以下的情况：

package main

import (
   "fmt"
   "sync"
)
func main() {
   numberFlag := 0
   wg := new(sync.WaitGroup)
   for i := 0; i < 200; i++ {
      wg.Add(1)
      go func() {
         defer wg.Done()
         numberFlag++
      }()
   }
   fmt.Println(numberFlag)
   wg.Wait()
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

每次执行后的计数器结果都是不同的，这是因为计数器本身是被不同的协程抢着+1，会产生多个协程同时拿到numberFlag=N的情况。为了避免这种资源竞争，要对资源进行加锁，使得同一时刻只有一个协程能对资源进行操控。

package main

import (
   "fmt"
   "sync"
)

func main() {
   numberFlag := 0
   myLock := make(chan struct{}, 1)
   wg := new(sync.WaitGroup)
   for i := 0; i < 200; i++ {
      wg.Add(1)
      go func() {

         defer func() {
            // 释放锁
            <-myLock
         }()
         defer wg.Done()
         // 抢锁
         myLock <- struct{}{}
         numberFlag++
      }()
   }
   wg.Wait()
   fmt.Println(numberFlag)
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

但是这种锁只能用于你自己的本地服务，一旦出现多服务，比如分布式，微服务，这样的场景，这个锁就没啥用了，这就需要分布式锁。

关于分布式锁，一般的实现就是用redis或者zookeeper实现。redis比较方便的就是大部分的服务都会使用redis，无需额外安装依赖，而zookeeper普通服务用的并不多，即使是kafka也在新版放弃了zookeeper。

zookeeper最大的好处就是可以通过心跳检测客户端的情况，进而避免重复得锁的问题。

但是同时也产生了一些问题，这个心跳检测多久一次，在心跳检测的间隔如果出现了锁超时的问题怎么办，等等。

所以一些服务还是倾向于使用redis来实现分布式锁。

package main

import (
   "fmt"
   "github.com/gomodule/redigo/redis"
   "go-face-test/redisTest/redisOne/redisConn"
   "sync"
   "time"
)

func main() {

   // 分布式锁
   var LockName = "lockLock"
   // 十秒过期时间
   var ExpirationTime = 10
   wg := new(sync.WaitGroup)
   wg.Add(2)
   // 起两个协程来模拟分布式服务的抢占
   go handleBusiness(LockName, ExpirationTime, "A", wg)
   go handleBusiness(LockName, ExpirationTime, "B", wg)
   wg.Wait()
}
func handleBusiness(lockName string, ExpTime int, nowGroName string, wg *sync.WaitGroup) {

   // One服务获取锁是否存在
   c := redisConn.Get()
   defer c.Close()

   for {
      isKeyExists, err := redis.Bool(c.Do("EXISTS", lockName))
      if err != nil {
         fmt.Println("err while checking keys:", err)
      } else {
         fmt.Println(isKeyExists)
      }
      if isKeyExists {
         // 存在这把锁，开始自旋等待
         fmt.Println("当前协程为: " + nowGroName + " 没抢到锁……")
         //休息1s
         time.Sleep(time.Second)
      } else {
         // 设置一把锁
         // 值为1，过期时间为10秒
         reply, err := c.Do("SET", lockName, 2, "EX", ExpTime, "NX")
         // 抢占失败
         if reply == nil {
            fmt.Println("当前协程为: " + nowGroName + " 抢占锁失败")
            continue
         }
         // 开始业务处理
         fmt.Println("当前协程为: " + nowGroName + " 啊啊啊啊。这是一个业务处理，预计处理时间为 3s 。处理开始........")
         fmt.Println("当前协程为: " + nowGroName + " 距离处理完成还有---3s" + time.Now().Format("2006-01-02 15:04:05"))
         time.Sleep(time.Second)
         fmt.Println("当前协程为: " + nowGroName + " 距离处理完成还有---2s" + time.Now().Format("2006-01-02 15:04:05"))
         time.Sleep(time.Second)
         fmt.Println("当前协程为: " + nowGroName + " 距离处理完成还有---1s" + time.Now().Format("2006-01-02 15:04:05"))
         time.Sleep(time.Second)

         //业务结束,释放锁
         _, err = c.Do("DEL", lockName)
         if err != nil {
            fmt.Println("err while deleting:", err)
         }
         break
      }
   }
   wg.Done()

}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70

但是这个锁明显有问题:


第一，当A服务（本案例中其实是协程模拟的）拿到锁之后，处理超时了，锁还没有释放，就已经过期，过期后B服务就抢到了锁，此时AB均认为自己拿到了锁


第二，A服务按理说只能去掉自己的服务加上的锁，如果不止是有AB两个服务，有更多的服务，那么A如果出现处理较慢，锁超时后，B服务抢到锁，A又处理完成所有的事释放了锁，那其实是释放掉了B的锁。也就是说，释放锁的时候也必须判断是否是自己的锁


那么就得用redis的lua来保证原子性

package main

import (
   "fmt"
   "github.com/gomodule/redigo/redis"
   "go-face-test/redisTest/redisTwo/redisConn"
   "log"
   "math/rand"
   "strconv"
   "sync"
   "time"
)

var letterRunes = []rune("abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ")
var lockCommand = `if redis.call("GET", KEYS[1]) == ARGV[1] then
    redis.call("SET", KEYS[1], ARGV[1], "PX", ARGV[2] , "NX")
    return "OK"
else
    return redis.call("SET", KEYS[1], ARGV[1], "NX", "PX", ARGV[2])
end`
var delCommand = `if redis.call("GET", KEYS[1]) == ARGV[1] then
    return redis.call("DEL", KEYS[1])
else
    return 0
end`

func main() {

   // 分布式锁
   var LockName = "lockLock"
   // 十秒过期时间
   var ExpirationTime = 3
   wg := new(sync.WaitGroup)
   wg.Add(2)
   // 起两个协程来模拟分布式服务的抢占
   go handleBusiness(LockName, ExpirationTime, "A", wg)
   go handleBusiness(LockName, ExpirationTime, "B", wg)
   wg.Wait()
}
func init() {
   rand.Seed(time.Now().UnixNano())
}
func handleBusiness(lockName string, ExpTime int, nowGroName string, wg *sync.WaitGroup) {

   // One服务获取锁是否存在
   c := redisConn.Get()
   defer c.Close()

   for {
      isKeyExists, err := redis.Bool(c.Do("EXISTS", lockName))
      if err != nil {
         fmt.Println("err while checking keys:", err)
      } else {
         fmt.Println(isKeyExists)
      }
      if isKeyExists {
         // 存在这把锁，开始自旋等待
         fmt.Println("当前协程为: " + nowGroName + " 没抢到锁……")
         //休息1s
         time.Sleep(time.Second)
      } else {
         // 设置一把锁
         // 锁的值是根据当前服务名称和时间来的
         lockFlag, lockValue, _ := getLock(lockName, nowGroName, ExpTime, c)
         // 抢占失败
         if !lockFlag {
            fmt.Println("当前协程为: " + nowGroName + " 抢占锁失败")
            continue
         }
         // 开始业务处理
         fmt.Println("当前协程为: " + nowGroName + " 啊啊啊啊。这是一个业务处理，预计处理时间为 " + strconv.Itoa(ExpTime) + "s 。处理开始........")
         for i := ExpTime - 1; i > 0; i-- {
            fmt.Println("当前协程为: " + nowGroName + " 距离处理完成还有---" + strconv.Itoa(i) + "s " + time.Now().Format("2006-01-02 15:04:05"))
            time.Sleep(time.Second)
         }

         //业务结束,释放锁
         lockDelFlag, _ := delLock(lockName, lockValue, c)
         //获取当前锁的值
         if lockDelFlag {
            fmt.Println("释放锁成功")
         } else {
            fmt.Println("这个锁不是你的，或者这个锁已经超时")
         }
         break
      }
   }
   wg.Done()
}

// 获得唯一锁的值
func getLockOnlyValue(nowGroName string) string {
   nano := strconv.FormatInt(time.Now().UnixNano(), 10)
   return nowGroName + "_" + nano + "_" + RandStringRunes(6)
}

// 获得一个锁
func getLock(LockName, nowGroName string, timeOut int, conn redis.Conn) (bool, string, error) {
   myLockValue := getLockOnlyValue(nowGroName)
   lua := redis.NewScript(1, lockCommand)
   resp, err := lua.Do(conn, LockName, myLockValue, strconv.Itoa(timeOut*1000))
   if err != nil {
      log.Fatal(LockName, err)
      return false, "", err
   } else if resp == nil {
      return false, "", nil
   }
   s, ok := resp.(string)
   if !ok {
      return false, "", nil
   }
   if s != "OK" {
      return false, "", nil
   }
   return true, myLockValue, nil
}

// 删除一个锁
func delLock(LockName, LockeValue string, conn redis.Conn) (bool, error) {
   lua := redis.NewScript(1, delCommand)
   resp, err := lua.Do(conn, LockName, LockeValue)
   if err != nil {
      return false, err
   }
   reply, ok := resp.(int64)
   if !ok {
      return false, nil
   }
   return reply == 1, nil
}
func RandStringRunes(n int) string {
   b := make([]rune, n)
   for i := range b {
      b[i] = letterRunes[rand.Intn(len(letterRunes))]
   }
   return string(b)
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137

​