汤不热吧在分布式系统中，多个服务实例同时访问共享资源时，传统的单机锁机制（如Java的synchronized、Python的threading.Lock）已经无法满足需求。Redis凭借其高性能和丰富的数据结构，成为实现分布式锁的首选方案。然而，一个看似简单的分布式锁，背后隐藏着大量的工程细节和陷阱。本文将从最基础的实现讲起，逐步深入到Redlock算法，帮助你掌握Redis分布式锁的核心原理和最佳实践。
一、为什么需要分布式锁
想象一个典型的电商场景：用户抢购一件限量商品，库存数量为1。在单机环境下，我们可以用互斥锁保证同一时刻只有一个线程能执行扣减库存的操作。但在微服务架构下，订单服务可能部署了多个实例，每个实例都有自己的进程空间，传统的进程内锁无法跨实例生效。
如果不对并发访问做控制，可能出现以下问题：
- 超卖问题：多个实例同时读取库存为1，都认为可以下单，导致卖出超过实际库存的订单
- 数据不一致：多个实例同时修改同一份数据，最终结果与预期不符
- 重复执行：定时任务在多个实例上同时触发，导致业务逻辑执行多次
分布式锁的核心目标非常简单：在整个分布式系统中，同一时刻只能有一个客户端持有锁。这个看似简单的需求，要在分布式环境下可靠地实现，需要考虑网络分区、进程崩溃、锁超时等诸多异常情况。
二、最基础的实现：SETNX + EXPIRE
Redis提供了SETNX命令（SET if Not eXists），只有在key不存在时才能设置成功，这天然适合用来实现锁的互斥语义。最基础的实现方式如下：
import redis
import time
client = redis.Redis(host='localhost', port=6379, db=0)
def acquire_lock(lock_name, timeout=10):
"""获取分布式锁，timeout为锁的过期时间（秒）"""
lock_key = f"lock:{lock_name}"
identifier = str(time.time()) # 唯一标识，用于安全释放锁
# SET key value NX EX seconds
result = client.set(lock_key.identifier, nx=True, ex=timeout)
if result:
return identifier
return None
def release_lock(lock_name, identifier):
"""释放分布式锁"""
lock_key = f"lock:{lock_name}"
# 使用Lua脚本保证原子性
lua_script = """
if redis.call('get', KEYS[1]) == ARGV[1] then
return redis.call('del', KEYS[1])
else
return 0
end
"""
return client.eval(lua_script, 1, lock_key, identifier)这段代码有三个关键设计：
- SET NX EX原子操作：使用
SET key value NX EX timeout一条命令同时完成”不存在则设置”和”设置过期时间”两个操作。早期的做法是先执行SETNX再执行EXPIRE，两条命令之间如果客户端崩溃，锁就不会过期，造成死锁。 - 唯一标识符：每个客户端获取锁时生成一个唯一标识（通常用UUID或时间戳），释放锁时需要验证标识符匹配才删除，防止误删其他客户端的锁。
- Lua脚本释放：释放锁时必须先比较值再删除，这两步需要用Lua脚本保证原子性。如果用GET + DEL两条命令，中间可能被其他客户端抢先获取锁。
三、锁的续期：Watchdog机制
使用固定过期时间的锁存在一个经典问题：如果业务逻辑执行时间超过了锁的过期时间，锁会自动释放，其他客户端就能获取到锁，导致并发问题。
Redisson（Java Redis客户端）提出了Watchdog（看门狗）机制来解决这个问题。核心思路是在后台启动一个定时任务，每隔一段时间检查锁是否仍然被持有，如果是则自动续期。
import threading
import uuid
import redis
class RedisDistributedLock:
def __init__(self, client, lock_name, expire=30):
self.client = client
self.lock_key = f"lock:{lock_name}"
self.identifier = str(uuid.uuid4())
self.expire = expire
self._watchdog_thread = None
self._stop_watchdog = threading.Event()
def acquire(self, blocking=True, timeout=None):
"""获取锁，支持阻塞等待"""
start = time.time()
while True:
if self.client.set(self.lock_key, self.identifier, nx=True, ex=self.expire):
self._start_watchdog()
return True
if not blocking:
return False
if timeout and (time.time() - start) >= timeout:
return False
time.sleep(0.1)
def release(self):
"""释放锁并停止看门狗"""
self._stop_watchdog.set()
self._release_lock()
def _release_lock(self):
lua_script = """
if redis.call('get', KEYS[1]) == ARGV[1] then
return redis.call('del', KEYS[1])
else
return 0
end
"""
return self.client.eval(lua_script, 1, self.lock_key, self.identifier)
def _start_watchdog(self):
"""启动看门狗线程，定期续期"""
def watchdog_loop():
interval = self.expire // 3 # 每隔过期时间的1/3续期一次
while not self._stop_watchdog.wait(timeout=interval):
# 续期：只有锁仍然属于自己时才续期
lua_script = """
if redis.call('get', KEYS[1]) == ARGV[1] then
return redis.call('expire', KEYS[1], ARGV[2])
else
return 0
end
"""
result = self.client.eval(
lua_script, 1, self.lock_key,
self.identifier, self.expire
)
if not result:
break # 锁已不属于自己，停止续期
self._watchdog_thread = threading.Thread(
target=watchdog_loop, daemon=True
)
self._watchdog_thread.start()Watchdog机制的几个要点：
- 续期间隔通常设为过期时间的1/3，保证在锁到期前至少能续期一次
- 续期必须是原子操作：用Lua脚本先检查锁的持有者再续期，避免给别人的锁续期
- 优雅退出：释放锁时要停止看门狗线程，否则可能给一个已经释放的锁续期
- 守护线程：使用daemon线程，进程退出时自动清理
四、可重入锁的实现
在复杂的业务场景中，同一个线程可能需要多次获取同一把锁（比如递归调用或方法嵌套）。如果锁不可重入，第二次获取就会死锁。Redis实现可重入锁需要使用Hash结构来记录锁的重入次数：
class ReentrantRedisLock:
"""基于Redis Hash的可重入分布式锁"""
ACQUIRE_SCRIPT = """
local key = KEYS[1]
local field = ARGV[1]
local ttl = tonumber(ARGV[2])
if redis.call('hexists', key, field) == 1 then
redis.call('hincrby', key, field, 1)
redis.call('expire', key, ttl)
return 1
end
if redis.call('hlen', key) == 0 then
redis.call('hincrby', key, field, 1)
redis.call('expire', key, ttl)
return 1
end
return 0
"""
RELEASE_SCRIPT = """
local key = KEYS[1]
local field = ARGV[1]
if redis.call('hexists', key, field) == 0 then
return 0
end
local count = redis.call('hincrby', key, field, -1)
if count == 0 then
redis.call('hdel', key, field)
end
if redis.call('hlen', key) == 0 then
redis.call('del', key)
end
return 1
"""
def __init__(self, client, lock_name, expire=30):
self.client = client
self.lock_key = f"lock:reentrant:{lock_name}"
self.field = f"{socket.gethostname()}:{threading.current_thread().ident}"
self.expire = expire
self._acquire_script = self.client.register_script(self.ACQUIRE_SCRIPT)
self._release_script = self.client.register_script(self.RELEASE_SCRIPT)
def acquire(self):
return bool(self._acquire_script(
keys=[self.lock_key],
args=[self.field, self.expire]
))
def release(self):
return bool(self._release_script(
keys=[self.lock_key],
args=[self.field]
))可重入锁的Hash结构设计如下：锁的key是Redis中的一个Hash，field是客户端标识（通常由主机名+线程ID组成），value是重入计数。每次获取锁时计数+1，释放时计数-1，计数降到0时才真正删除锁。
五、Redis在实际生产环境中的常见问题
5.1 缓存与数据库的一致性
Redis作为缓存层使用时，与数据库的数据一致性是最常被讨论的问题。常见的策略有Cache Aside Pattern：
def get_user(user_id):
# 1. 先查缓存
cache_key = f"user:{user_id}"
cached = redis_client.get(cache_key)
if cached:
return json.loads(cached)
# 2. 缓存未命中，查数据库
user = db.query("SELECT * FROM users WHERE id = %s", user_id)
if user:
# 3. 写入缓存，设置过期时间
redis_client.setex(cache_key, 3600, json.dumps(user))
return user
def update_user(user_id, data):
# 1. 先更新数据库
db.execute("UPDATE users SET ... WHERE id = %s", user_id)
# 2. 删除缓存（而不是更新缓存）
redis_client.delete(f"user:{user_id}")为什么是删除缓存而不是更新缓存？因为并发写入时，两个线程更新数据库的顺序和更新缓存的顺序可能不一致，导致缓存中存储了脏数据。删除缓存可以避免这个问题，下次读取时会重新从数据库加载。
5.2 大Key问题的处理
当Redis中存储的单个Key的Value过大时（比如一个Hash包含几百万个field，或一个List包含几百万个元素），会导致以下问题：
- 阻塞其他请求：Redis是单线程的，操作大Key会耗时较长，阻塞其他命令
- 内存不均衡：在集群模式下，大Key可能集中在某个节点上
- 网络带宽占用：传输大Value消耗大量带宽
- 主从同步延迟：同步大Key会占用大量时间和带宽
检测和处理大Key的方法：
# 使用redis-cli的--bigkeys扫描大Key
redis-cli --bigkeys
# 使用MEMORY USAGE查看单个Key的内存占用
redis-cli MEMORY USAGE mykey
# 使用SCAN遍历所有Key，配合DEBUG OBJECT查看大小
redis-cli --memkeys --memkeys-patterns '*'处理大Key的原则是拆分：将一个大Hash拆分为多个小Hash，按业务维度分片存储。例如存储用户行为日志时，按用户ID取模分散到多个Key中。
5.3 热Key问题
热Key是指某个Key在短时间内被大量访问。在集群环境下，热Key会导致某个节点的QPS远高于其他节点，造成负载不均。解决方案包括：
- 本地缓存：在应用层增加本地缓存（如Caffeine、guava-cache），减少对Redis的请求
- Key复制：将热Key复制多份，读取时随机选择其中一个，分散压力
- 读写分离：读请求路由到从节点，写请求走主节点
六、Redis持久化：RDB与AOF的对比
Redis提供两种持久化方式，各有优劣。了解它们的原理和适用场景，是做好Redis运维的基础。
| 对比维度 | RDB（快照） | AOF（追加日志） |
|---|---|---|
| 持久化方式 | 定期生成内存数据的二进制快照文件 | 将每条写命令追加到日志文件 |
| 数据安全性 | 可能丢失最后一次快照后的数据 | 根据fsync策略最多丢失1秒数据 |
| 文件大小 | 紧凑，体积小 | 较大，但支持重写压缩 |
| 恢复速度 | 快，直接加载二进制数据 | 慢，需要重放所有写命令 |
| 对性能影响 | fork子进程时可能短暂阻塞 | fsync策略影响写入性能 |
| 适用场景 | 灾备恢复、数据备份 | 数据安全性要求高的场景 |
在生产环境中，通常建议同时开启RDB和AOF。AOF用于保证数据安全（最多丢失1秒数据），RDB用于快速恢复和冷备份。Redis重启时优先使用AOF文件恢复数据，因为AOF的数据更完整。
七、Redlock算法：Redis分布式锁的最佳实践
前面讨论的单节点Redis锁存在一个致命问题：如果Redis主节点故障，在主从切换过程中，锁可能丢失。假设客户端A在主节点上获取锁，主节点宕机后从节点提升为新主，此时客户端B也能在新主上获取同一把锁，锁的互斥性被打破。
为了解决这个问题，Redis作者Antirez提出了Redlock算法。核心思想是使用多个独立的Redis实例（推荐5个），客户端需要在大多数实例上成功获取锁才算获取成功：
import time
import uuid
import redis
class Redlock:
"""Redlock分布式锁实现"""
def __init__(self, nodes, ttl=30000):
"""
nodes: Redis实例列表
ttl: 锁的过期时间（毫秒）
"""
self.nodes = [
redis.Redis(host=n['host'], port=n['port'], db=0,
socket_timeout=0.2)
for n in nodes
]
self.ttl = ttl
self.quorum = len(nodes) // 2 + 1 # 多数派
def acquire(self, resource):
"""获取分布式锁"""
identifier = str(uuid.uuid4())
retry = 3
for _ in range(retry):
# 统计成功获取锁的实例数
n = 0
start_time = int(time.time() * 1000)
for node in self.nodes:
if self._acquire_instance(node, resource, identifier):
n += 1
# 计算获取锁花费的时间
elapsed = int(time.time() * 1000) - start_time
validity = self.ttl - elapsed - 2 # 2ms的时钟漂移补偿
# 需要在多数派实例上成功，且锁的有效期大于0
if n >= self.quorum and validity > 0:
return {'resource': resource, 'identifier': identifier,
'validity': validity}
else:
# 获取失败，释放所有已获取的锁
for node in self.nodes:
self._release_instance(node, resource, identifier)
time.sleep(0.2)
return None
def _acquire_instance(self, node, resource, identifier):
try:
return node.set(f"lock:{resource}", identifier,
nx=True, px=self.ttl)
except redis.RedisError:
return False
def release(self, lock):
"""释放分布式锁"""
for node in self.nodes:
self._release_instance(node, lock['resource'], lock['identifier'])
def _release_instance(self, node, resource, identifier):
lua = """
if redis.call('get', KEYS[1]) == ARGV[1] then
return redis.call('del', KEYS[1])
end
return 0
"""
try:
node.eval(lua, 1, f"lock:{resource}", identifier)
except redis.RedisError:
passRedlock算法的争议
值得一提的是，Redlock算法在分布式系统领域引发了广泛讨论。Martin Kleppmann在其文章《How to do distributed locking》中指出Redlock存在时钟漂移、GC停顿等问题，建议使用fencing token（隔离令牌）来保证安全性。Antirez随后进行了回应和解释。
在实际生产中，如果你的分布式锁用于保护非常关键的操作（如金融交易），建议：
- 使用ZooKeeper或etcd等基于共识协议的方案，它们在理论上更安全
- 或者在Redlock基础上增加fencing token机制，确保即使锁失效也不会导致数据错误
- 如果只是用于优化性能（如防空击），单节点Redis锁已经足够
Redis分布式锁的核心价值在于它简单、高效、适用面广。理解了本文介绍的各种场景和注意事项，你就能在实际项目中做出正确的技术选型和实现决策。记住：没有完美的分布式锁方案，只有在特定约束下最合适的方案。