6.2 分布式锁

在单机程序并发或并行修改全局变量时,需要对修改行为加锁以创造临界区。为什么需要加锁呢?可以看看这段代码:

  1. package main
  2. import (
  3. "sync"
  4. )
  5. // 全局变量
  6. var counter int
  7. func main() {
  8. var wg sync.WaitGroup
  9. for i := 0; i < 1000; i++ {
  10. wg.Add(1)
  11. go func() {
  12. defer wg.Done()
  13. counter++
  14. }()
  15. }
  16. wg.Wait()
  17. println(counter)
  18. }

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

  1. ❯❯❯ go run local_lock.go
  2. 945
  3. ❯❯❯ go run local_lock.go
  4. 937
  5. ❯❯❯ go run local_lock.go
  6. 959

6.2.1 进程内加锁

想要得到正确的结果的话,要把对 counter 的操作代码部分加上锁:

  1. // ... 省略之前部分
  2. var wg sync.WaitGroup
  3. var l sync.Mutex
  4. for i := 0; i < 1000; i++ {
  5. wg.Add(1)
  6. go func() {
  7. defer wg.Done()
  8. l.Lock()
  9. counter++
  10. l.Unlock()
  11. }()
  12. }
  13. wg.Wait()
  14. println(counter)
  15. // ... 省略之后部分

这样就可以稳定地得到计算结果了:

  1. ❯❯❯ go run local_lock.go
  2. 1000

6.2.2 trylock

  1. package main
  2. import (
  3. "sync"
  4. )
  5. // Lock try lock
  6. type Lock struct {
  7. c chan struct{}
  8. }
  9. // NewLock generate a try lock
  10. func NewLock() Lock {
  11. var l Lock
  12. l.c = make(chan struct{}, 1)
  13. l.c <- struct{}{}
  14. return l
  15. }
  16. // Lock try lock, return lock result
  17. func (l Lock) Lock() bool {
  18. lockResult := false
  19. select {
  20. case <-l.c:
  21. lockResult = true
  22. default:
  23. }
  24. return lockResult
  25. }
  26. // Unlock , Unlock the try lock
  27. func (l Lock) Unlock() {
  28. l.c <- struct{}{}
  29. }
  30. var counter int
  31. func main() {
  32. var l = NewLock()
  33. var wg sync.WaitGroup
  34. for i := 0; i < 10; i++ {
  35. wg.Add(1)
  36. go func() {
  37. defer wg.Done()
  38. if !l.Lock() {
  39. // log error
  40. println("lock failed")
  41. return
  42. }
  43. counter++
  44. println("current counter", counter)
  45. l.Unlock()
  46. }()
  47. }
  48. wg.Wait()
  49. }

因为我们的逻辑限定每个 goroutine 只有成功执行了 Lock 才会继续执行后续逻辑,因此在 Unlock 时可以保证 Lock struct 中的 channel 一定是空,从而不会阻塞,也不会失败。

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

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

6.2.3 基于 redis 的 setnx

  1. package main
  2. import (
  3. "fmt"
  4. "sync"
  5. "time"
  6. "github.com/go-redis/redis"
  7. )
  8. func incr() {
  9. client := redis.NewClient(&redis.Options{
  10. Addr: "localhost:6379",
  11. Password: "", // no password set
  12. DB: 0, // use default DB
  13. })
  14. var lockKey = "counter_lock"
  15. var counterKey = "counter"
  16. // lock
  17. resp := client.SetNX(lockKey, 1, time.Second*5)
  18. lockSuccess, err := resp.Result()
  19. if err != nil || !lockSuccess {
  20. fmt.Println(err, "lock result: ", lockSuccess)
  21. return
  22. }
  23. // counter ++
  24. getResp := client.Get(counterKey)
  25. cntValue, err := getResp.Int64()
  26. if err == nil {
  27. cntValue++
  28. resp := client.Set(counterKey, cntValue, 0)
  29. _, err := resp.Result()
  30. if err != nil {
  31. // log err
  32. println("set value error!")
  33. }
  34. }
  35. println("current counter is ", cntValue)
  36. delResp := client.Del(lockKey)
  37. unlockSuccess, err := delResp.Result()
  38. if err == nil && unlockSuccess > 0 {
  39. println("unlock success!")
  40. } else {
  41. println("unlock failed", err)
  42. }
  43. }
  44. func main() {
  45. var wg sync.WaitGroup
  46. for i := 0; i < 10; i++ {
  47. wg.Add(1)
  48. go func() {
  49. defer wg.Done()
  50. incr()
  51. }()
  52. }
  53. wg.Wait()
  54. }

看看运行结果:

  1. ❯❯❯ go run redis_setnx.go
  2. <nil> lock result: false
  3. <nil> lock result: false
  4. <nil> lock result: false
  5. <nil> lock result: false
  6. <nil> lock result: false
  7. <nil> lock result: false
  8. <nil> lock result: false
  9. <nil> lock result: false
  10. <nil> lock result: false
  11. current counter is 2028
  12. unlock success!

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

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

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

6.2.4 基于 zk

  1. package main
  2. import (
  3. "time"
  4. "github.com/samuel/go-zookeeper/zk"
  5. )
  6. func main() {
  7. c, _, err := zk.Connect([]string{"127.0.0.1"}, time.Second) //*10)
  8. if err != nil {
  9. panic(err)
  10. }
  11. l := zk.NewLock(c, "/lock", zk.WorldACL(zk.PermAll))
  12. err = l.Lock()
  13. if err != nil {
  14. panic(err)
  15. }
  16. println("lock succ, do your business logic")
  17. time.Sleep(time.Second * 10)
  18. // do some thing
  19. l.Unlock()
  20. println("unlock succ, finish business logic")
  21. }

基于 zk 的锁与基于 redis 的锁的不同之处在于 Lock 成功之前会一直阻塞,这与我们单机场景中的 mutex.Lock 很相似。

其原理也是基于临时 sequence 节点和 watch api,例如我们这里使用的是 /lock 节点。Lock 会在该节点下的节点列表中插入自己的值,只要节点下的子节点发生变化,就会通知所有 watch 该节点的程序。这时候程序会检查当前节点下最小的子节点的 id 是否与自己的一致。如果一致,说明加锁成功了。

这种分布式的阻塞锁比较适合分布式任务调度场景,但不适合高频次持锁时间短的抢锁场景。按照 Google 的 chubby 论文里的阐述,基于强一致协议的锁适用于 粗粒度 的加锁操作。这里的粗粒度指锁占用时间较长。我们在使用时也应思考在自己的业务场景中使用是否合适。

6.2.5 基于 etcd

  1. package main
  2. import (
  3. "log"
  4. "github.com/zieckey/etcdsync"
  5. )
  6. func main() {
  7. m, err := etcdsync.New("/lock", 10, []string{"http://127.0.0.1:2379"})
  8. if m == nil || err != nil {
  9. log.Printf("etcdsync.New failed")
  10. return
  11. }
  12. err = m.Lock()
  13. if err != nil {
  14. log.Printf("etcdsync.Lock failed")
  15. return
  16. }
  17. log.Printf("etcdsync.Lock OK")
  18. log.Printf("Get the lock. Do something here.")
  19. err = m.Unlock()
  20. if err != nil {
  21. log.Printf("etcdsync.Unlock failed")
  22. } else {
  23. log.Printf("etcdsync.Unlock OK")
  24. }
  25. }

etcd 中没有像 zookeeper 那样的 sequence 节点。所以其锁实现和基于 zookeeper 实现的有所不同。在上述示例代码中使用的 etcdsync 的 Lock 流程是:

  1. 先检查 /lock 路径下是否有值,如果有值,说明锁已经被别人抢了
  2. 如果没有值,那么写入自己的值。写入成功返回,说明加锁成功。写入时如果节点被其它节点写入过了,那么会导致加锁失败,这时候到 3
  3. watch /lock 下的事件,此时陷入阻塞
  4. /lock 路径下发生事件时,当前进程被唤醒。检查发生的事件是否是删除事件(说明锁被持有者主动 unlock),或者过期事件(说明锁过期失效)。如果是的话,那么回到 1,走抢锁流程。

6.2.6 redlock

  1. package main
  2. import (
  3. "fmt"
  4. "time"
  5. "github.com/garyburd/redigo/redis"
  6. "gopkg.in/redsync.v1"
  7. )
  8. func newPool(server string) *redis.Pool {
  9. return &redis.Pool{
  10. MaxIdle: 3,
  11. IdleTimeout: 240 * time.Second,
  12. Dial: func() (redis.Conn, error) {
  13. c, err := redis.Dial("tcp", server)
  14. if err != nil {
  15. return nil, err
  16. }
  17. return c, err
  18. },
  19. TestOnBorrow: func(c redis.Conn, t time.Time) error {
  20. _, err := c.Do("PING")
  21. return err
  22. },
  23. }
  24. }
  25. func newPools(servers []string) []redsync.Pool {
  26. pools := []redsync.Pool{}
  27. for _, server := range servers {
  28. pool := newPool(server)
  29. pools = append(pools, pool)
  30. }
  31. return pools
  32. }
  33. func main() {
  34. pools := newPools([]string{"127.0.0.1:6379", "127.0.0.1:6378", "127.0.0.1:6377"})
  35. rs := redsync.New(pools)
  36. m := rs.NewMutex("/lock")
  37. err := m.Lock()
  38. if err != nil {
  39. panic(err)
  40. }
  41. fmt.Println("lock success")
  42. unlockRes := m.Unlock()
  43. fmt.Println("unlock result: ", unlockRes)
  44. }

redlock 也是一种阻塞锁,单个节点操作对应的是 set nx px 命令,超过半数节点返回成功时,就认为加锁成功。

关于 redlock 的设计曾经在社区引起一场口水战,分布式专家各抒己见。不过这个不是我们要讨论的内容,相关链接在参考资料中给出。

6.2.7 如何选择

业务还在单机就可以搞定的量级时,那么按照需求使用任意的单机锁方案就可以。

如果发展到了分布式服务阶段,但业务规模不大,比如 qps < 1000,使用哪种锁方案都差不多。如果公司内已有可以使用的 zk/etcd/redis 集群,那么就尽量在不引入新的技术栈的情况下满足业务需求。

业务发展到一定量级的话,就需要从多方面来考虑了。首先是你的锁是否在任何恶劣的条件下都不允许数据丢失,如果不允许,那么就不要使用 redis 的 setnx 的简单锁。

如果要使用 redlock,那么要考虑你们公司 redis 的集群方案,是否可以直接把对应的 redis 的实例的 ip+port 暴露给开发人员。如果不可以,那也没法用。

对锁数据的可靠性要求极高的话,那只能使用 etcd 或者 zk 这种通过一致性协议保证数据可靠性的锁方案。但可靠的背面往往都是较低的吞吐量和较高的延迟。需要根据业务的量级对其进行压力测试,以确保分布式锁所使用的 etcd/zk 集群可以承受得住实际的业务请求压力。需要注意的是,etcd 和 zk 集群是没有办法通过增加节点来提高其性能的。要对其进行横向扩展,只能增加搭建多个集群来支持更多的请求。这会进一步提高对运维和监控的要求。多个集群可能需要引入 proxy,没有 proxy 那就需要业务去根据某个业务 id 来做 sharding。如果业务已经上线的情况下做扩展,还要考虑数据的动态迁移。这些都不是容易的事情。

在选择具体的方案时,还是需要多加思考,对风险早做预估。