使用redis设计一个简单的分布式锁

Posted 2020-12-29 lonelyxmas

原文:使用redis设计一个简单的分布式锁

最近看了有关redis的一些东西，了解了redis的一下命令，就记录一下：

redis中的setnx命令：

关于redis的操作命令，我们一般会使用set，get等一系列操作，数据结构也有很多，这里我们使用最简单的string来存储锁。

redis下提供一个setnx命令用来将key值设为value，类似于set功能，但是它和set是有区别的，在于后面的nx，setnx是SET if Not eXists。就是：当且仅当key值不存在的时候，将该key值设置为value。

也就是说使用setnx加入元素的时候，当key值存在的时候，是没有办法加入内容的。

加锁：

下面将使用python控制redis，python控制redis的方式和命令一样，有一个setnx(key, value)方法，通过这个方法可以实现setnx命令的效果。

首先连接redis：

文件connect.py

1 import redis
2 
3 def connect_redis():
4     return redis.Redis(host=‘127.0.0.1‘, port=6379)

然后就可以使用setnx加锁了，key对应的value中需要填入相应的值，这里设置Value为一个uuid值。获取到uuid之后，将key和value填入redis，其他的客户端想要访问并获取到锁，也使用setnx方式，key值只要相同就好。那么代码如下：

文件operate.py

1 # 加锁的过程
2 def acquire_lock(conn, lockname):
3     identifier = str(uuid.uuid4())
4     conn.setnx(‘lock:‘ + lockname, identifier): 

这样就通过setnx将key值写入了。但是，这样显然是不合理的，客户端可以等待一会再次获取锁，这样，不至于每次请求都会出现问题。那可以设置30秒的时间，让程序在30秒内不停的尝试获取锁，知道30秒的时间过了或者由其他客户端释放了锁。

所以加锁可以变为如下：

 1 # 加锁的过程
 2 def acquire_lock(conn, lockname, args, acquite_timeout = 30):
 3     identifier = str(uuid.uuid4())
 4     end = time.time() + acquite_timeout
 5     while time.time() < end:
 6         # 这里尝试取得锁 setnx 设置-如果不存在的时候才会set
 7         if conn.setnx(‘lock:‘ + lockname, identifier): 
 8             # 获得锁之后输出获得锁的‘进程’号
 9             print(‘获得锁:进程‘+ str(args))
10             return identifier
11     return False

这样就可以通过setnx加锁了，这个加锁的过程实际上就是在redis中存入了一个值，之后当其他的客户端再次想要通过这个key加入这个值的时候，发现这个key已经存在就不往进写值了，但是在这30秒内还会不断尝试的去获取锁，也就是不断的尝试写入这个值，一旦key被删除——也就是锁被释放，则该客户端就可以竞争获取锁——进程写入这个值。这种方式就像是操作系统中的自旋锁。

自旋锁

这里先简单介绍一下自旋锁。

和自旋锁对应的还有一种锁，叫做对于互斥锁。

互斥锁：如果资源已经被占用，资源申请者只能进入睡眠状态。

自旋锁：自旋锁不会引起调用者睡眠，如果自旋锁已经被别的执行单元保持，调用者就一直循环在那里看是否该自旋锁的保持者已经释放了锁，"自旋"一词就是因此而得名。

释放锁：

既然锁可以添加，那么也就可以释放。释放锁实际上就是将redis中的数据删除。这里可以使用redis提供的事务流水线去执行。为了保障在执行的时候确确实实释放的所示没有问题的。

代码如下，调用delete()方法——del命令删除redis中这个key的元素。

 1 # 释放锁
 2 def release_lock(conn, lockname, identifier):
 3     pipe = conn.pipeline(True)
 4     lockname = ‘lock:‘ + lockname
 5     while True:
 6         try:
 7             pipe.watch(lockname)
 8             identifier_real = pipe.get(lockname)
 9             if identifier_real == identifier:
10                 pipe.multi()
11                 pipe.delete(lockname)
12                 pipe.execute()
13                 return True;
14             pipe.unwatch()
15             break
16         except redis.exceptions.WatchError:
17             pass
18     return False

这里就遇到了python3中的一个坑，获取到的数据为byte类型，所以identifier_real这个变量实际上是这个字符串之前加了个b的，例如b‘xxxx‘，所以这和传入的identifier的类型为string，这样比较当然会出现Fasle，所以我们在这里将这个字符串类型转码：

1 pipe.get(lockname).decode()

这样才是整整的字符串类型的字符串，最终的代码如下：

 1 def release_lock(conn, lockname, identifier):
 2     pipe = conn.pipeline(True)
 3     lockname = ‘lock:‘ + lockname
 4     while True:
 5         try:
 6             pipe.watch(lockname)
 7             identifier_real = pipe.get(lockname).decode()
 8             if identifier_real == identifier:
 9                 pipe.multi()
10                 pipe.delete(lockname)
11                 pipe.execute()
12                 return True;
13             pipe.unwatch()
14             break
15         except redis.exceptions.WatchError:
16             pass
17     return False

执行：

为了验证这个分布式锁的正确性，可以写一个测试的方法进行测试，先去获取锁，如果获取到之后，则sleep三秒，等待3秒之后，执行释放锁。

模拟过程入下：

文件operate.py

1 # 模拟加锁解锁的过程
2 def exec_test(conn, lockname, args):
3     id = acquire_lock(conn, lockname, args)
4     if id != False:
5         print(id)
6         time.sleep(3)
7         release_lock(conn, lockname, id)

这样我们就可以使用多进程并发访问的方式进行对锁进行抢占和释放：

使用9个进程进行测试，操作方式如下：

1 from connect import connect_redis
2 from operate import acquire_lock
3 from multiprocessing import Process
4 from operate import exec_test
5 
6 if __name__ == ‘__main__‘:
7     redis = connect_redis()
8     for i in range(0, 9):
9         Process(target = exec_test, args = (redis, ‘test‘, i)).start()

执行结果如下所示：

注：以上python运行环境为python3.6