项目地址: https://github.com/liwook/Redislock
1.支持阻塞式等待获取锁
之前的是只尝试获取一次锁，要是获取失败就不再尝试了。现在修改为支持阻塞式等待获取锁。
添加LockOptions结构体
添加option.go文件。
在LockOptions中
- isBlock表示是否是阻塞模式
- blockWaitingTime是获取key的阻塞超时时间
- expire表示key的过期时间(之前是在结构体RedisLock，现在保存在LockOptions中)
const (
DefaultExpireTime = 20 * time.Second
DefaultBlockWaitingTime = 8 * time.Second
)
type LockOptions struct {
isBlock bool
blockWaitingTime time.Duration
expire time.Duration
}下面是LockOption的设置方法。
//option.go
type LockOptionFunc func(*LockOptions)
// 设置阻塞等待
func WithBlock() LockOptionFunc {
return func(option *LockOptions) {
option.isBlock = true
}
}
//设置阻塞等待时间的上限
func WithBlockWaiting(waiting time.Duration) LockOptionFunc {
return func(option *LockOptions) {
option.blockWaitingTime = waiting
}
}
//设置续期的时长，也是key过期的时长
func WithExpire(exprie time.Duration) LockOptionFunc {
return func(option *LockOptions) {
option.expire = exprie
}
}
func setLock(o *LockOptions) {
if o.isBlock && o.blockWaitingTime <= 0 {
//默认阻塞等待时间上限是8
o.blockWaitingTime = 8
}
if o.expire == 0 {
o.expire = DefaultExpireTime
}
}修改创建锁的代码
//该结构体添加了LockOptions，去掉了expire成员
type RedisLock struct {
LockOptions
key string
Id string //锁的标识
redisCli *redis.Client
}
func NewRedisLock(cli *redis.Client, key string, opts ...LockOptionFunc) *RedisLock {
id := strings.Join(strings.Split(uuid.New().String(), "-"), "")
lock := RedisLock{
key: key,
Id: id,
redisCli: cli,
}
//执行一些配置操作
for _, optFunc := range opts {
optFunc(&lock.LockOptions)
}
setLock(&lock.LockOptions)
return &lock
}
//用法
lock := redislock.NewRedisLock(client, key, redislock.WithBlock(), redislock.WithBlockWaiting(10*time.Second))
//之前的写法
// func NewRedisLock(cli *redis.Client, key string) *RedisLock {
// id := strings.Join(strings.Split(uuid.New().String(), "-"), "")
// return &RedisLock{
// key: key,
// expire: defaultExpireTime,
// Id: id,
// redisCli: cli,
// }
// }加锁
加锁主要分成了3步：
- 不管是不是阻塞的，都先尝试获取一次锁tryLock()
- 非阻塞加锁失败的话，就直接返回错误
- 之后基于阻塞模式轮询去获取锁
func (lock *RedisLock) Lock() (bool, error) {
//不管是否是阻塞的，都是要先获取一次锁
success, err := lock.tryLock()
if success && err == nil {
return success, err
}
//非阻塞加锁失败的话，直接返回错误
if !lock.isBlock {
return false, err
}
//基于阻塞模式轮询去获取锁
return lock.blockingLock()
}
func (lock *RedisLock) tryLock() (bool, error) {
return lock.redisCli.SetNX(lock.key, lock.Id, lock.expire).Result()
}阻塞模式中使用了定时器轮询去获取锁。
func (lock *RedisLock) blockingLock() (bool, error) {
timeoutCh := time.After(lock.blockWaitingTime)
//轮询ticker，定时器, 100ms循环一次去获取锁
ticker := time.NewTicker(100 * time.Millisecond)
defer ticker.Stop()
for {
select {
case <-timeoutCh:
return false, fmt.Errorf("block waiting timeout,err:%w", ErrLockAcquiredByOthers)
case <-ticker.C:
success, err := lock.tryLock() //尝试获取锁
if success && err == nil {
return success, nil
}
}
}
}测试使用
这样lock先后顺序可以获得锁了。
func main() {
testBlockingLock()
}
func testBlockingLock() {
client := NewClient()
defer client.Close()
val, _ := client.Ping().Result()
fmt.Println(val)
key := "blockLock"
lock := redislock.NewRedisLock(client, key, redislock.WithBlock(), redislock.WithBlockWaiting(15*time.Second))
var wg sync.WaitGroup
wg.Add(1)
go func() {
//尝试获取锁
if success, err := lock.Lock(); success && err == nil {
fmt.Println("go BLOCKlock get..")
time.Sleep(4 * time.Second)
lock.Unlock()
}
wg.Done()
}()
//尝试获取锁
if success, err := lock.Lock(); success && err == nil {
fmt.Println("BLOCKlock get...")
time.Sleep(7 * time.Second)
lock.Unlock()
}
wg.Wait()
}2.锁续期的看门狗实现
这里仍然存在一个问题：当锁的持有者任务未完成，但是锁的有效期已过，虽然持有者此时仍可以完成任务，并且也不会误删其他持有者的锁，但是此时可能会存在多个执行者同时执行临界区代码，使得数据的一致性难以保证，造成意外的后果，分布式锁就失去了意义。
因此，需要一个锁的自动续期机制，分布式锁框架Redission中就有这么一个看门狗，专门为将要到期的锁进行续期。这里我们也来实现一个简单的看门狗。
在LockOptions添加关于锁续期和看门狗标识
const (
// 默认的分布式锁过期时间,也是默认的续期时长
DefaultExpireTime = 20 * time.Second
// 看门狗工作时间间隙
WatchDogWorkStepTime = 10 * time.Second
..........
)
type LockOptions struct {
................
//强调，expire是key的过期时长，也是要进行续期时的续期时长
expire time.Duration
wathchDogMode bool
}下面是关于锁续期和看门狗标识的设置方法。
//设置续期的时长，也是key过期的时长，(在支持阻塞式等待获取锁的时候已展示过)
func WithExpire(exprie time.Duration) LockOptionFunc {
return func(option *LockOptions) {
option.expire = exprie
}
}
func setLock(o *LockOptions) {
if o.isBlock && o.blockWaitingTime <= 0 {
//没有设置默认阻塞时间就使用默认阻塞时长
o.blockWaitingTime = DefaultBlockWaitingTime
}
if o.watchDogWorkStepTime == 0 {
o.watchDogWorkStepTime = DefaultWatchDogWorkStepTime
}
//简单起见，就设置是开启看门狗模式
o.wathchDogMode = true
if o.expire == 0 {
o.expire = DefaultExpireTime
}
//比较续期时长和看门狗工作时间间隔
if o.expire <= o.watchDogWorkStepTime {
o.watchDogWorkStepTime = o.expire - 2
}
}添加watchDog方法
在watchDog内部开启新协程执行runWatchDog。把context.WithCancel的结果赋值给结构体RedisLock的stopDog，到时解锁的时候就可以调用RedisLock.stopDog就可以停止看门狗，回收看门狗协程。协程中调用runWatchDog方法。
type RedisLock struct {
....................
// 停止看门狗
stopDog context.CancelFunc //通过context.CancelFunc去停止看门狗
}
func (lock *RedisLock) watchDog() {
if !lock.wathchDogMode {
return
}
var ctx context.Context
ctx, lock.stopDog = context.WithCancel(context.Background())
//启动看门狗
go func() {
lock.runWatchDog(ctx)
}()
}runWatchDog方法中使用了go语言标准库中的Ticker实现定时查看锁是否过期。
在select 语句中，每隔WatchDogWorkStepTime秒就会触发一次 ticker进行续期，将key的过期时间重置。注意，这里也是使用Lua脚本封装了确认锁与锁续期的操作来用于原子化，以防止误续期了其他持有者的锁。
func (lock *RedisLock) runWatchDog(ctx context.Context) error {
//开启一个定时器
ticker := time.NewTicker(lock.watchDogWorkStepTime)
defer ticker.Stop()
script := redis.NewScript(LauCheckThenExpire)
for {
select {
case <-ticker.C:
result, err := script.Run(lock.redisCli, []string{lock.key}, lock.Id, lock.expire+3).Result()
if err != nil {
return err
}
if ret, _ := result.(int64); ret != 1 {
return errors.New("can not expire lock without ownership of lock")
}
case <-ctx.Done():
return nil
}
}
}加锁时刻
相比起之前的，主要是添加了开头的defer函数。只要最终是获取了锁，就执行watchDog()。
func (lock *RedisLock) Lock() (success bool, err error) {
defer func() {
if success && err == nil {
lock.watchDog()
}
}()
//不管是否是阻塞的，都是要先获取一次锁
success, err = lock.tryLock()
if success && err == nil {
return success, err
}
//非阻塞加锁失败的话，直接返回错误
if !lock.isBlock {
return false, err
}
//基于阻塞模式轮询去获取锁
success, err = lock.blockingLock()
return
}解锁时刻
相比之前的，也是添加了defer函数。这里就是用lock.stopDog()来停止看门狗，也规避潜在的协程泄漏问题.
func (lock *RedisLock) Unlock() error {
defer func() {
//停止看门狗
if lock.stopDog != nil {
lock.stopDog()
}
}()
script := redis.NewScript(LauCheckAndDelete)
result, err := script.Run(lock.redisCli, []string{lock.key}, lock.Id).Int64()
if err != nil {
return err
}
if result != 1 {
return errors.New("can not unlock without ownership of lock")
}
return nil
}3.RedLock实现
为什么需要RedLock
redis 的容错机制：为避免单点故障引起数据丢失问题，redis 会基于主从复制的方式实现数据备份增加服务的容错性.
以哨兵机制为例，哨兵会持续监听 master 节点的健康状况，倘若 master 节点发生故障，哨兵会负责扶持 slave 节点上位成为 master，以保证整个集群能够正常对外提供服务。
在分布式系统存在一个经典的 CAP 理论。
-
C：consistency，一致性
-
A：availability，可用性
-
P：Partition tolerance，分区容错性
redis 走的是 AP 路线，为了保证服务的可用性和吞吐量，redis 在进行数据的主从同步时，采用的是异步执行机制。
我们试想一种场景：
-
时刻1：使用方 A 在 redis master 节点加锁成功，完成了锁数据的写入操作
-
时刻2：redis master 宕机了，锁数据还没来得及同步到 slave 节点
-
时刻3：未同步到锁数据的 slave 节点被哨兵升级为新的 master
-
时刻4：使用方 B 前来取锁，由于新 master 中确实锁数据，所以使用方 B 加锁成功
这个时候可以使用redis红锁（redlock，全称 redis distribution lock)。redLock 的策略是通过增加锁的数量并基于多数派准则来解决这个问题。
保证在 RedLock 下所有 redis 节点中达到半数以上节点可用时，整个红锁就能够正常提供服务。
规则的具体细节：
- 获取当前的时间(毫秒)
- 使用相同的key和随机值在N个master上获取锁。这里获取锁的尝试时间要远远小于锁的超时时间，是为了防止某个master挂了之后我们还在不停获取锁，导致被阻塞时间过长。比如:该锁20s过期，三个节点加锁花了21秒，那就是加锁失败。
- 在大多数master上获取到了锁，并且中的获取时间小于锁的过期时间的情况下，才会被认为锁获取成功。
- 如果锁获取成功，那锁的超时时间 = 最初的锁超时时间 - 获取锁的总耗时时间。
- 如果锁获取失败，不管是因为获取成功的master的个数没有过半，还是因为获取锁的耗时超过了锁的过期时间，都会将已经设置了该key的master上的把该key删除。
添加关于红锁的Option和结构体RedLock
添加结构体RedLockOptions。其内包括了单个节点的请求耗时的超时时间singleNodeTimeout和整个红锁的过期时间。
//option.go
//红锁的操作
type RedLockOptionFunc func(*RedLockOptions)
type RedLockOptions struct {
singleNodeTimeout time.Duration //单个节点的请求耗时的超时时间
exprie time.Duration //整个红锁的过期时间
}
func WithSingleNodeTimeout(singleNodeTimeout time.Duration) RedLockOptionFunc {
return func(opt *RedLockOptions) {
opt.singleNodeTimeout = singleNodeTimeout
}
}
func WithRedLockExpire(expire time.Duration) RedLockOptionFunc {
return func(opt *RedLockOptions) {
opt.exprie = expire
}
}
func setRedLock(opt *RedLockOptions) {
if opt.singleNodeTimeout <= 0 {
opt.singleNodeTimeout = DefaultSingleLockTimeout
}
if opt.exprie <= 0 {
opt.exprie = DefaultExpireTime
}
}新添redlock.go文件。添加结构体RedLock。
其是对多个节点进行加锁，锁的数量会增多，所以 RedLock中会存有*RedisLock的数组，还有RedLock的一些选项配置。
//redlock.go
//单个节点的请求锁的耗时时间上限
const DefaultSingleLockTimeout = 50 * time.Millisecond
type RedLock struct {
locks []*RedisLock
RedLockOptions
}创建RedLock
因为是多个节点了，就会有多个节点的client，可以在option.go文件中创建结构体SingleNode，其中存有redis的地址和密码。
//option.go
type SingleNode struct {
Address string //redis的地址
Password string //redis的密码
}创建红锁主要分成4个步骤：
- 判断节点的个数，小于3个无意义
- 进行红锁的配置设置option
- 判断所有节点累计的加锁超时时间是否小于设定的分布式锁过期时间的1/10，这点是对应 获取锁的尝试时间要远远小于锁的超时时间。(不一定要1/10,可以自己设置)
- 对所有节点进行连接，并创建每个节点的redislock，并赋值给红锁的成员locks
func NewRedLock(key string, nodes []*SingleNode, opts ...RedLockOptionFunc) (*RedLock, error) {
//步骤1 ，节点个数<3，没有意义
if len(nodes) < 3 {
return nil, errors.New("the number of node is less than 3")
}
//步骤2
lock := RedLock{}
for _, opt := range opts {
opt(&lock.RedLockOptions)
}
setRedLock(&lock.RedLockOptions)
//步骤3
if lock.exprie > 0 && time.Duration(len(nodes))*lock.singleNodeTimeout*10 > lock.exprie {
// 要求所有节点累计的超时阈值要小于分布式锁过期时间的十分之一
return nil, errors.New("expire thresholds of single node is too long")
}
//步骤4
lock.locks = make([]*RedisLock, 0, len(nodes))
for _, node := range nodes {
client := redis.NewClient(&redis.Options{
Addr: node.Address,
Password: node.Password,
})
lock.locks = append(lock.locks, NewRedisLock(client, key, WithExpireSeconds(lock.exprie)))
}
return &lock, nil
}加锁
对每个node进行加锁。并且对在singleNodeTimeout耗时时间内的加锁成功的锁进行计数。
要是加锁成功的个数超过一半，那即是加锁成功。
func (r *RedLock) Lock() (bool, error) {
//成功加锁的个数
successNum := 0
//对每个node尝试加锁
for _, lock := range r.locks {
startTime := time.Now()
success, err := lock.Lock()
cost := time.Since(startTime)
if err == nil && success && cost <= r.singleNodeTimeout {
successNum++
}
}
if successNum < (len(r.locks)>>1)+1 {
return false, errors.New("lock failed,lock nodes are Not enough for half")
}
return true, nil
}解锁
需要对所有节点进行解锁。其解锁是使用了(RedisLock).Unlock()。
// 解锁，需对所有节点解锁
func (r *RedLock) Unlock() error {
var allErr error
for _, lock := range r.locks {
if err := lock.Unlock(); err != nil {
if allErr == nil {
allErr = err
}
}
}
return allErr
}测试使用
func main() {
testReadLock()
}
func testReadLock() {
nodes := getNodes()
key := "redLock"
redLock, err := redislock.NewRedLock(key, nodes, redislock.WithRedLockExpire(10*time.Second), redislock.WithSingleNodeTimeout(100*time.Millisecond))
if err != nil {
return
}
var wg sync.WaitGroup
wg.Add(1)
go func() {
//lock1尝试获取锁
if success, err := redLock.Lock(); success && err == nil {
fmt.Println("go redLock get..")
time.Sleep(4 * time.Second)
redLock.Unlock()
}
wg.Done()
}()
//lock2尝试获取锁
if success, err := redLock.Lock(); success && err == nil {
fmt.Println("redLock get...")
time.Sleep(7 * time.Second)
redLock.Unlock()
}
wg.Wait()
}
func getNodes() []*redislock.SingleNode {
//三个节点
addr1 := "127.0.0.1:10000"
passwd1 := "okredis"
addr2 := "127.0.0.1:10001"
passwd2 := "okredis"
addr3 := "127.0.0.1:10002"
passwd3 := "okredis"
return []*redislock.SingleNode{
{
Address: addr1,
Password: passwd1,
},
{
Address: addr2,
Password: passwd2,
},
{
Address: addr3,
Password: passwd3,
},
}
}还是会存在的问题
在5台机器中(都是master）,在代码中依次对这5台机器去加锁，只有成功的机器数大于一半就算加锁成功，其他机器也就没必要再去操作了，相反，如果大于一半的机器失败了，就算失败，其他机器也就没必要再去操作了。
这时一样会出问题。
- 线程A要加锁，对1,2,3,4,5这5个实例进行加锁。1,2,3成功，4,5加锁超时，那这时有三个master加锁成功，已超过一半，即是最终加锁成功了。
- 而这时节点3挂了。很快运维人员把一个新节点顶替已挂的节点3。
- 在新节点还没有该锁key时候，线程B来获取该锁，这时节点3,4,5就获取锁成功，也因为成功个数超过一半，也即是获取锁成功。这时就有两个线程同时获取同一把锁。
所以说红锁也是不能完全解决所有问题的。
Redis 官网关于红锁的描述，你能看到著名的关于红锁的神仙打架事件。即 Martin Kleppmann 和 Antirez 的 RedLock 辩论。一个是很有资历的分布式架构师，一个是 Redis 之父。
所以，使用红锁还是需要慎重。而且本文章实现的红锁是比较简单的，还有很多细节没有考虑到的。
![[lesson16]类的真正形态](https://img-blog.csdnimg.cn/direct/62f31b48b994474492be6d9fc3d01785.png#pic_center)
