python下使用redis分布式锁
1.什么场景需要分布式锁?
我们在写业务逻辑的时候，如果多个线程同时访问某个共享变量，一般是对变量进行上锁或者使用 queue.Queue() 实现，以做到线程安全保证数据不被污染。
在单机部署的情况下这样做完全没问题，但是随着业务规模的发展，某些单机部署的系统需要进化成分布式集群系统，需要部署到多台服务器上去做负载均衡，这就会使得以前的并发控制策略失效，这个时候就需要一种跨机器的互斥机制来控制共享资源的访问，以保证线程安全，这个时候就需要使用到分布式锁。
2.分布式锁应该具备的条件：
① 在分布式系统环境下，同一时间只能有一个客户端能持有锁，且保证加锁和解锁的必须是同一个客户端
② 为了防止死锁，需要具备锁失效机制，即使一个客户端在持有锁期间崩溃没有主动释放锁，也能保证后续其他客户端能正常加锁
③ 具备非堵塞特性，没有获取到锁则直接返回获取锁失败。
3.redis 分布式锁的原理
redis 中的 SETNX 命令可以实现「key不存在才插入」:
如果 key 不存在，则显示插入成功，用来表示加锁成功
如果 key 存在，则会显示插入失败，用来表示加锁失败
通过 setnx 设置分布式锁，拿到这个锁的线程就可以执行业务代码，没有拿到的则只能进行等待，执行完业务代码后的线程需要通过 del key 释放锁，让其他线程能够重新获取，这样就能实现在分布式系统并发情况下，始终只有一个线程在执行业务代码。
4.demo
'''
pip install redis
'''
import time
import uuid
import redis
from threading import Thread
# redis 存字符串返回的是byte, 指定 decode_responses=True 可以解决
pool = redis.ConnectionPool(host="127.0.0.1", port=6379, socket_connect_timeout=3, decode_responses=True)
redis_cli = redis.Redis(connection_pool=pool)
# 加锁
def acquire_lock(lock_name, acquire_timeout=4, lock_timeout=7):
"""
param lock_name: 锁名称
param acquire_timeout: 客户端获取锁的超时时间
param lock_timeout: 锁过期时间, 超过这个时间锁自动释放
"""
identifier = str(uuid.uuid4())
end_time = time.time() + acquire_timeout # 客户端获取锁的结束时间
while time.time() <= end_time:
# setnx(key, value) 只有 key 不存在情况下将 key 设置为 value 返回 True
# 若 key 存在则不做任何动作,返回 False
if redis_cli.setnx(lock_name, identifier):
redis_cli.expire(lock_name, lock_timeout) # 设置锁的过期时间，防止线程获取锁后崩溃导致死锁
return identifier # 返回锁唯一标识
elif redis_cli.ttl(lock_name) == -1: # 当锁未被设置过期时间时，重新设置其过期时间
redis_cli.expire(lock_name, lock_timeout)
time.sleep(0.001)
return False # 获取超时返回 False
# 释放锁
def release_lock(lock_name, identifier):
"""
param lock_name: 锁名称
param identifier: 锁标识
"""
# 解锁操作需要在一个 redis 事务中进行，python 中 redis 事务通过 pipeline 封装实现
with redis_cli.pipeline() as pipe:
while True:
try:
# 使用 WATCH 监听锁，如果删除过程中锁自动失效又被其他客户端拿到，即锁标识被其他客户端修改
# 此时设置了 WATCH 事务就不会再执行，这样就不会出现删除了其他客户端锁的情况
pipe.watch(lock_name)
id = pipe.get(lock_name)
if id and id == identifier: # 判断解锁与加锁线程是否一致
pipe.multi() # 开启事务
pipe.delete(lock_name) # 标识相同，在事务中删除锁
pipe.execute() # 执行EXEC命令后自动执行UNWATCH
return True
pipe.unwatch()
break
except redis.WatchError:
pass
return False
# 发售商品件数
goods = 10
def exec_test(thread_name):
identifier = acquire_lock('jaye')
if identifier: # 如果获取到锁,则执行业务逻辑
print(f'{thread_name}获取 redis 分布式锁成功！')
time.sleep(3) # 模拟业务耗时
global goods
goods = goods - 1
print(f'------------------------{thread_name}抢到了东西------------------------------')
res = release_lock('jaye', identifier) # 处理完之后释放锁
print(f'{thread_name} 释放状态: {res}')
else:
print(f'{thread_name}获取 redis 分布式锁失败, 其他线程正在使用')
if __name__ == '__main__':
for i in range(1000):
t_name = f'thread_{i}'
t = Thread(target=exec_test, args=(t_name,))
t.start()
time.sleep(10)
print('还剩下{}'.format(goods))
5.注意
'''
ps：redis主从模式中数据的复制是异步的，如果在主节点获取到锁后，没来得及同步到从节点就宕机了，哨兵重新选举新的主节点后依然可以获取锁，也就导致多个服务可以同时拿到锁操作资源，这种情况下可以使用Redlock。
'''