最近研究了一下redis的分布式锁，总体来说，虽然有一些缺点，但是对于小规模的并发还是比较实用的。
先说下为什么用redis作为锁（个人主观感受）
1、redis使用内存存储，加锁和释放锁都比较快
2、redis是单线程的程序，所有的操作都是串行化执行的，不会有几个client同时触发的情况
3、可以用lua脚本模拟其他数据库的事物，可将多个连续操作封装成一个原子操作，提交给redis
接下来还是上代码（一直觉得代码注释的形式胜过一行行口述）：
# 单例的元类，保证被构造的类传入相同的参数时只实例化1次
class SingleCache(type):
# 初始化一个缓存字典，用于缓存instance
def __init__(cls, name, bases, dct: dict):
super().__init__(name, bases, dct)
cls.__cache_dict = {}
# 连接池绑定为类属性，供实例访问
cls.pool = redis.ConnectionPool(
host='127.0.0.1', port='6379', password='',
max_connections=20, decode_responses=True
)
# 类的实例化时调用
def __call__(cls, *args, **kwargs):
cache_tuple = args + tuple(sorted(kwargs.items()))
if cache_tuple not in cls.__cache_dict:
cls.__cache_dict[cache_tuple] = type.__call__(cls, *args, **kwargs)
return cls.__cache_dict[cache_tuple]
class RedisLockUtil(metaclass=SingleCache):
def __init__(self):
# 复用类属性连接池（听说这个连接池源码有些问题，但是使用中还好，没发现太大问题，可能场景比较有限）
self.pool = self.pool
# 从连接池中获取一个连接
def get_conn(self):
r = redis.Redis(connection_pool=self.pool)
return r
# 给资源加锁，resource_id为资源id，request_id为当前操作者的唯一id，会把resource_id作为key，request_id及一些简单的用户信息作为value存入redis
def add_lock(self, resource_id: str, request_id: str, ex=30):
# 因为预设了获取锁成功后，返回值是1，失败后返回是，所以
assert(request_id != '1'), f'1为脚本默认返回值，所以请求id不能为1'
user_info = {'user_id': '1', 'user_name': '张三'}
value = f'{request_id}|{json.dumps(user_info, ensure_ascii=False)}'
lua_script_list = [
# 查看request_id是否已经存在
'local _value = redis.call("get", KEYS[1]);',
# 如果不存在，就设置，并配置过期时间，默认30秒，然后返回字符串1
'if(_value==false)',
'then',
'redis.call("set",KEYS[1],ARGV[1]);',
'redis.call("expire",KEYS[1],ARGV[3]);',
'return "1";',
'end;',
# 如果存在，则key的结构是这样的：request_id|user_info，竖线分隔，找到竖线索引，取出request_id和user_info
'local split_index = string.find(_value, "|");',
'local request_id = string.sub(_value,0,split_index-1);',
# 如果请求id和当前value中的一致，则可以获取锁，返回字符串1，否则不能获取锁
'if(request_id == ARGV[2])',
'then',
'return "1";',
'end;',
# 如果request_id不同，说明获取锁失败，此时把占用当前锁的用户信息返回
'local user_info = string.sub(_value,split_index+1);',
'return user_info;'
]
# lua脚本从列表转到字符串
lua_script = '\n'.join(lua_script_list)
conn = self.get_conn()
# 注册lua脚本
_script = conn.register_script(lua_script)
# 传入所需参数，keys对应的是lua脚本中的KEYS，args对应的是lua脚本中的ARGV，注意下标索引是从1开始，而非0
result = _script(keys=[resource_id], args=[value, request_id, ex])
# 如果没有返回1，说明获取锁失败，抛出异常
if result != "1":
user_info = json.loads(result)
msg = f'当前资源id：{resource_id}已被占用，占用人信息:user_id:%(user_id)s,user_name:%(user_name)s' % user_info
raise Exception(msg)
print(f'加锁成功，resource_id:{resource_id},request_id:{request_id}')
# 释放锁
def release_lock(self, resource_id: str, request_id: str):
lua_script_list = [
# 检查resource_id是否存在
'local _value = redis.call("get", KEYS[1]);',
# resource_id不存在的返回字符串0
'if(_value==false)',
'then',
'return "0";',
'end;',
# 如果resource_id存在，则比较request_id是否相同，如果相同，则可以释放
'local split_index = string.find(_value, "|");',
'local request_id = string.sub(_value,0,split_index-1);',
'if(request_id == ARGV[1])',
'then',
'redis.call("del", KEYS[1]);',
'return "1";',
'end;',
# 如果request_id不同，释放锁失败，返回当前request_id
'return ARGV[1];'
]
# lua脚本从列表转到字符串
lua_script = '\n'.join(lua_script_list)
conn = self.get_conn()
# 注册lua脚本
_script = conn.register_script(lua_script)
# 传入所需参数，keys对应的是lua脚本中的KEYS，args对应的是lua脚本中的ARGV，注意下标索引是从1开始，而非0
result = _script(keys=[resource_id], args=[request_id])
# 如果返回0，说明资源已经不存在，无需释放锁
if result == '0':
print(f"资源id为:{resource_id}的锁已不存在")
# 如果返回1，说明已经成功释放锁
elif result == '1':
print(f"资源id为:{resource_id}的锁已释放")
# 如果返回其他，说明request_id不符，无法释放锁
else:
print(f"解锁请求id为:{request_id},与加锁id:{result}不同，无法释放锁")
msg = '资源释放失败'
raise Exception(msg)
以上就是完整代码，可根据实际场景做一些增加或是删除，但基本思想已经表达的足够清晰了，接下来我们做几个测试，看看实际执行效果：
rlu = RedisLockUtil()
rlu.add_lock(resource_id='lirui123', request_id='123')
# 加锁成功，resource_id:lirui123,request_id:123
rlu.release_lock(resource_id='lirui123', request_id='123')
# 资源id为:lirui123的锁已释放
rlu.add_lock(resource_id='lirui123', request_id='123')
# 加锁成功，resource_id:lirui123,request_id:123
rlu.release_lock(resource_id='lirui123', request_id='1234')
# Traceback (most recent call last):
# File "redis_lock2.py", line 125, in <module>
# rlu.release_lock(resource_id='lirui123', request_id='1234')
# File "redis_lock2.py", line 116, in release_lock
# raise Exception(msg)
# Exception: 资源释放失败
rlu.release_lock(resource_id='lirui123', request_id='123')
# 资源id为:lirui123的锁已释放
rlu.add_lock(resource_id='lirui123', request_id='123', ex=5)
# 加锁成功，resource_id:lirui123,request_id:123
import time
time.sleep(5)
rlu.release_lock(resource_id='lirui123', request_id='123')
# 资源id为:lirui123的锁已不存在
rlu.add_lock(resource_id='lirui123', request_id='123')
# 加锁成功，resource_id:lirui123,request_id:123
rlu.add_lock(resource_id='lirui123', request_id='1234')
# File "redis_lock2.py", line 122, in <module>
# rlu.add_lock(resource_id='lirui123', request_id='1234')
# File "redis_lock2.py", line 75, in add_lock
# raise Exception(msg)
# Exception: 当前资源id：lirui123已被占用，占用人信息:user_id:1,user_name:张三
以上，基本覆盖了平时用锁的基本场景，基本都可以cover住。不过对于这种用法，还存在一个问题：当redis是集群时，如果某个node挂掉，因为redis数据是异步同步策略，有一段时间数据会不完全，这会让锁异常或者失效。
然而这种属于小概率事件了，对于我这种懒人来说，可以暂时不用考虑了~~
