前言

随着互联网技术的不断发展，用户量的不断增加，越来越多的业务场景需要用到分布式系统。而在分布式系统中访问共享资源就需要一种互斥机制，来防止彼此之间的互相干扰，以保证一致性，这个时候就需要使用分布式锁。

业界常用解决方案基于 MySql 等数据库的唯一索引

基于 ZooKeeper 临时有序节点

基于 Redis 的 NX EX 参数

本文主要讲解基于 Redis 实现的分布式锁

分布式锁的特点互斥性。在任意时刻，只有一个客户端能持有锁

锁超时。即使一个客户端持有锁的期间崩溃而没有主动释放锁，也需要保证后续其他客户端能够加锁成功

加锁和解锁必须是同一个客户端，客户端自己不能把别人加的锁给释放了。

实现

版本一

代码

# -*- coding: utf-8 -*-

# @DateTime : 2020/3/9 15:36

# @Author : woodenrobot

import uuid

import math

import time

from redis import WatchError

def acquire_lock_with_timeout(conn, lock_name, acquire_timeout=3, lock_timeout=2):

"""基于 Redis 实现的分布式锁:param conn: Redis 连接:param lock_name: 锁的名称:param acquire_timeout: 获取锁的超时时间，默认 3 秒:param lock_timeout: 锁的超时时间，默认 2 秒:return:"""

identifier = str(uuid.uuid4())

lockname = f'lock:{lock_name}'

lock_timeout = int(math.ceil(lock_timeout))

end = time.time() + acquire_timeout

while time.time() < end:

# 如果不存在这个锁则加锁并设置过期时间，避免死锁

if conn.setnx(lockname, identifier):

conn.expire(lockname, lock_timeout)

return identifier

# 如果存在锁，且这个锁没有过期时间则为其设置过期时间，避免死锁

elif conn.ttl(lockname) == -1:

conn.expire(lockname, lock_timeout)

time.sleep(0.001)

return False

def release_lock(conn, lockname, identifier):

"""释放锁:param conn: Redis 连接:param lockname: 锁的名称:param identifier: 锁的标识:return:"""

# python 中 redis 事务是通过pipeline的封装实现的

with conn.pipeline() as pipe:

lockname = 'lock:' + lockname

while True:

try:

# watch 锁, multi 后如果该 key 被其他客户端改变, 事务操作会抛出 WatchError 异常

pipe.watch(lockname)

iden = pipe.get(lockname)

if iden and iden.decode('utf-8') == identifier:

# 事务开始

pipe.multi()

pipe.delete(lockname)

pipe.execute()

return True

pipe.unwatch()

break

except WatchError:

pass

return False

加锁过程首先需要为锁生成一个唯一的标识，这里使用 uuid;

然后使用 setnx 设置锁，如果该锁名之前不存在其他客户端的锁则加锁成功，接着设置锁的过期时间防止发生死锁并返回锁的唯一标示；

如果设置失败先判断一下锁名所在的锁是否有过期时间，因为 setnx 和 expire 两个命令执行不是原子性的，可能会出现加锁成功但是设置超时时间失败出现死锁。如果不存在就给锁重新设置过期时间，存在就不断循环知道加锁时间超时加锁失败。

解锁过程首先整个解锁操作需要在一个 Redis 的事务中进行；

使用 watch 监听锁，防止解锁时出现删除其他人的锁；

查询锁名所在的标识是否与本次解锁的标识相同；

如果相同则在事务中删除这个锁，如果删除过程中锁自动失效过期又被其他客户端拿到，因为设置了 watch 就会删除失败，这样就不会出现删除了其他客户端锁的情况。

版本二

如果你使用的 Redis 版本大于等于 2.6.12 版本，加锁的过程就可以进行简化。因为这个版本以后的 Redis set 操作支持 EX 和 NX 参数，是一个原子性的操作。EX seconds ： 将键的过期时间设置为 seconds 秒。 执行 SET key value EX seconds 的效果等同于执行 SETEX key seconds value 。

NX ： 只在键不存在时， 才对键进行设置操作。 执行 SET key value NX 的效果等同于执行 SETNX key value 。

# -*- coding: utf-8 -*-

# @DateTime : 2020/3/9 15:36

# @Author : woodenrobot

import uuid

import math

import time

from redis import WatchError

def acquire_lock_with_timeout(conn, lock_name, acquire_timeout=3, lock_timeout=2):

"""基于 Redis 实现的分布式锁:param conn: Redis 连接:param lock_name: 锁的名称:param acquire_timeout: 获取锁的超时时间，默认 3 秒:param lock_timeout: 锁的超时时间，默认 2 秒:return:"""

identifier = str(uuid.uuid4())

lockname = f'lock:{lock_name}'

lock_timeout = int(math.ceil(lock_timeout))

end = time.time() + acquire_timeout

while time.time() < end:

# 如果不存在这个锁则加锁并设置过期时间，避免死锁

if conn.set(lockname, identifier, ex=lock_timeout, nx=True):

return identifier

time.sleep(0.001)

return False

def release_lock(conn, lockname, identifier):

"""释放锁:param conn: Redis 连接:param lockname: 锁的名称:param identifier: 锁的标识:return:"""

# python中redis事务是通过pipeline的封装实现的

with conn.pipeline() as pipe:

lockname = 'lock:' + lockname

while True:

try:

# watch 锁, multi 后如果该 key 被其他客户端改变, 事务操作会抛出 WatchError 异常

pipe.watch(lockname)

iden = pipe.get(lockname)

if iden and iden.decode('utf-8') == identifier:

# 事务开始

pipe.multi()

pipe.delete(lockname)

pipe.execute()

return True

pipe.unwatch()

break

except WatchError:

pass

return False

版本三

可能你也发现了解锁过程在代码逻辑上稍微有点复杂，别着急，我们可以使用 Lua 脚本实现原子性操作从而简化解锁过程。

# -*- coding: utf-8 -*-

# @DateTime : 2020/3/9 15:36

# @Author : woodenrobot

import uuid

import math

import time

def acquire_lock_with_timeout(conn, lock_name, acquire_timeout=3, lock_timeout=2):

"""基于 Redis 实现的分布式锁:param conn: Redis 连接:param lock_name: 锁的名称:param acquire_timeout: 获取锁的超时时间，默认 3 秒:param lock_timeout: 锁的超时时间，默认 2 秒:return:"""

identifier = str(uuid.uuid4())

lockname = f'lock:{lock_name}'

lock_timeout = int(math.ceil(lock_timeout))

end = time.time() + acquire_timeout

while time.time() < end:

# 如果不存在这个锁则加锁并设置过期时间，避免死锁

if conn.set(lockname, identifier, ex=lock_timeout, nx=True):

return identifier

time.sleep(0.001)

return False

def release_lock(conn, lock_name, identifier):

"""释放锁:param conn: Redis 连接:param lockname: 锁的名称:param identifier: 锁的标识:return:"""

unlock_script = """if redis.call("get",KEYS[1]) == ARGV[1] thenreturn redis.call("del",KEYS[1])elsereturn 0end"""

lockname = f'lock:{lock_name}'

unlock = conn.register_script(unlock_script)

result = unlock(keys=[lockname], args=[identifier])

if result:

return True

else:

return False

后续

截至到目前，我们已经有较好的方法获取锁和释放锁。基于Redis单实例，假设这个单实例总是可用，这种方法已经足够安全。但是如果 Redis 主节点挂了就会出现一些问题，比如主节点加锁后没有同步到从节点，从节点升为主节点，就会出现锁的丢失。如果你想要使用更加安全的 Redis 分布式锁实现可以参考一下 Redlock 的实现。

参考

第一时间获取最新文章更新，欢迎订阅我的微信公众号：woodenrobot