Go语言实现Redis分布式锁

基于go-redis的设计与实现

本文将基于go语言，使用了一个常用的go Redis客户端 go-redis库 , 一步一步探索与实现一个简单的Redis分布式锁。

代码：https://github.com/liwook/Redislock

连接Redis


func NewClient() *redis.Client {return redis.NewClient(&redis.Options{Addr:     "127.0.0.1:6379",    //自己的redis实例的ip和portPassword: "",    //密码，有设置的话，就需要填写})
}func main() {client := NewClient()defer client.Close()val, _ := client.Ping().Result()    //测试pingfmt.Println(val)
}

1.基于 SETNX 的锁初步实现

SETNX 命令用于在Redis中设置某个不存在的键的值。如果该键不存在，则设置成功，如果该键存在，则设置失败，不作任何动作。基于此可以实现一种简单的抢占机制。

新建lock.go文件。创建lock结构体，添加加锁和解锁方法。

结构体RedisLock有成员key，过期时间expire，连接的redis客户端redisCli。

var (defaultExpireTime = 5 * time.Second
)type RedisLock struct {key      stringexpire   time.DurationredisCli *redis.Client
}func NewRedisLock(cli *redis.Client, key string) *RedisLock {return &RedisLock{key:      key,expire:   defaultExpireTime,redisCli: cli,}
}

加锁

func (lock *RedisLock) Lock() (bool, error) {return lock.redisCli.SetNX(lock.key, "111111", lock.expire).Result()
}

上面的加锁是一种简单的方法，非阻塞的，一有结果就直接返回，也不再二次尝试的。lock.redisCli.SetNX(lock.key, "111111", lock.expire) 这行代码本质上执行了如下Redis操作命令：

set key 111111 ex 5 nx

该命令为 my_lock 键以 NX 方式设置了值。

如果持有锁的进程万一挂了，那么该键将永远存在与Redis中，其他竞争者无法进行 SETNX 操作，形成死锁。为了防范这种情况发生，这里设置了过期时间为5s，这样即便持锁者挂了，锁在一定时间后依然后自动释放。这里整个 set 操作是原子性的，并对该操作的返回结果作了判断，如果成功设置，说明抢占锁成功，则函数返回，进入临界区可以继续执行下面的代码。

解锁

func (lock *RedisLock) Unlock() error {res, err := lock.redisCli.Del(lock.key).Result()if err != nil {return err}if res != 1 {return errors.New("can not unlock because del result not is one")}return nil
}

上述代码中，lock.redisCli.Del(lock.key)对Redis中的lock 键进行了删除操作，当删除后，其他竞争者才有机会对该键进行 SETNX操作。

测试使用

func main() {client := NewClient()defer client.Close()val, _ := client.Ping().Result()fmt.Println(val)key := "mylock"lock := redislock.NewRedisLock(client, key)var wg sync.WaitGroupwg.Add(1)go func() {//尝试获取锁if success, err := lock.Lock(); success && err == nil {fmt.Println("go lock get..")time.Sleep(4 * time.Second)lock.Unlock()}wg.Done()}()//尝试获取锁// time.Sleep(1 * time.Second)if success, err := lock.Lock(); success && err == nil {fmt.Println(" main lock get...")time.Sleep(7 * time.Second)lock.Unlock()}wg.Wait()
}

2.锁的防误删实现

上面的就使用Redis实现了一个简单的分布式锁。但会存在个问题，想象一个场景：这个键过期了，但是其持有者线程A仍未完成任务。但这时该键就已经没有，线程B就去获取锁，获取成功了。这时候线程A完成了任务，就去删除键。而这时键是被线程B持有的，而线程A却可以去删除，这就会出了问题。

所以，这里要解决的是只有自己才能删除自己创建的锁。为了解决这种问题，持有者可以给锁添加一个唯一标识，使之只能删除自己的锁。因此需要完善一下加解锁操作：

在结构体RedisLock中添加字段Id,这是唯一标识符，用uuid表示。

在创建锁时候，需要创建出uuid，并赋值给字段Id。

type RedisLock struct {key      stringexpire   time.DurationId       string //锁的标识，新添加的，也即是键的valueredisCli *redis.Client
}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,}
}

那么在加锁的时候，把lock.Id给value赋值。

func (lock *RedisLock) Lock() (bool, error) {return lock.redisCli.SetNX(lock.key, lock.Id, lock.expire).Result()
}//对比之前的
//func (lock *RedisLock) Lock() (bool, error) {
//	return lock.redisCli.SetNX(lock.key, "111111", lock.expire).Result()
//}

解锁的时候，需要先判断锁的唯一标识值是否是与当前拥有者相匹配，若匹配再进行删除。

// 锁的误删除实现
func (lock *RedisLock) Unlock() error {//获取锁并进行判断该锁是否是自己的val, err := lock.redisCli.Get(lock.key).Result()if err != nil {fmt.Println("lock not exit")return err}if val == "" || val != lock.Id {return errors.New("lock not belong to myself")}//进行删除锁res, err := lock.redisCli.Del(lock.key).Result()if err != nil {return err}if res != 1 {return errors.New("can not unlock because del result not is one")}return nil
}

3.解锁的原子化实现

上面的解锁操作中，仍然是存在一个问题的：在确认当前锁是自己的锁后，和删除锁之前，这个时间段，中途可能会进行阻塞，这个过程中，锁恰巧过期释放，且被其他竞争者抢占。那就有可能会删除了其他竞争者的锁。这是不妥的。

我们要把这两个操作变成原子操作，将整个解锁过程原子化，使得在解锁期间，其他竞争者的任何操作不能被Redis执行。

Redis中可以使用Lua脚本把一系列操作变成原子操作。

func (lock *RedisLock) Unlock() error {script := redis.NewScript(LauCheckAndDelete)res, err := script.Run(lock.redisCli, []string{lock.key}, lock.Id).Int64()if err != nil {return err}if res != 1 {return errors.New("can not unlock because del result not is one")}return nil
}//lua.go
const (LauCheckAndDelete = `if(redis.call('get',KEYS[1])==ARGV[1]) thenreturn redis.call('del',KEYS[1])elsereturn 0end`
)

确认锁与删除锁的整体操作进行了原子化，便可以防止上述存在的误删问题。