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,会产生多个协程同时拿到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)
}

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

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

zookeeper最大的好处就是可以通过心跳检测客户端的情况,进而避免重复得锁的问题。
但是同时也产生了一些问题,这个心跳检测多久一次,在心跳检测的间隔如果出现了锁超时的问题怎么办,等等。

使用redis来实现分布式锁

所以一些服务还是倾向于使用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()
}

但是这个锁明显有问题:

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

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

也就是说,释放锁的时候也必须判断是否是自己的锁

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

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)
}

以上就是go 分布式锁简单实现实例详解的详细内容,更多关于go 分布式锁的资料请关注编程宝库其它相关文章!

 1.如何构建应用框架一般来说构建应用框架包含3个部分:命令行参数解析配置文件解析应用的命令行框架:需要具备 Help 功能、需要能够解析命令行参数和配置文件、命令需要能够初 ...