Redis 分布式锁实现：Redlock 算法的原理与避坑指南
前言
分布式锁是分布式系统中保证资源互斥访问的核心机制。相比单机环境的线程锁，分布式锁需要在多进程、多机器之间提供一致性保证。Redis 凭借其单线程执行和丰富的原子命令，是实现分布式锁的热门选择。但 Redis 分布式锁有很多坑，错误的实现可能导致严重的数据问题。本文讲解分布式锁的完整实现与 Redlock 算法。
为什么需要分布式锁
在分布式系统中，多个进程可能同时访问共享资源（如库存扣减、余额操作、订单幂等性校验），如果没有互斥机制，会导致数据不一致。
典型问题场景：
- 库存超卖：多个请求同时读到库存=1，都执行扣减，导致库存变成负数
- 重复下单：用户快速点击两次，两个请求同时检查订单不存在，都创建了订单
- 任务重复执行：定时任务部署了多个实例，同一时间都有可能被触发
# 问题示例：没有锁的库存扣减
def deduct_stock_wrong(product_id, quantity):
product = db.query("SELECT stock FROM products WHERE id = %s", (product_id,))
if product['stock'] < quantity:
return False
db.execute(
"UPDATE products SET stock = stock - %s WHERE id = %s",
(quantity, product_id)
)
return True -- 并发时两个请求都会返回 True，造成超卖
单节点 Redis 分布式锁
正确的 Redis 分布式锁需要满足以下特性：
1. 互斥性：任意时刻只有一个客户端能持有锁
2. 无死锁：即使客户端崩溃，锁也能被正确释放（通过过期时间）
3. 可重入：同一客户端可以多次获取同一把锁
import redis
import time
import uuid
r = redis.Redis(host='localhost', port=6379, decode_responses=True)
class RedisLock:
def __init__(self, redis_client, key, expire_seconds=10):
self.redis = redis_client
self.key = f"lock:{key}"
self.expire_seconds = expire_seconds
self.value = str(uuid.uuid4())
def acquire(self, blocking=True, timeout=10):
if not blocking:
return bool(self.redis.set(self.key, self.value, nx=True, ex=self.expire_seconds))
start_time = time.time()
while time.time() - start_time < timeout:
if self.redis.set(self.key, self.value, nx=True, ex=self.expire_seconds):
return True
time.sleep(0.01)
return False
def release(self):
-- Lua 脚本保证原子性：检查 value 是否匹配然后删除
lua_script = """
if redis.call("get", KEYS[1]) == ARGV[1] then
return redis.call("del", KEYS[1])
else
return 0
end
"""
self.redis.eval(lua_script, 1, self.key, self.value)
def __enter__(self):
if not self.acquire():
raise RuntimeError(f"Failed to acquire lock: {self.key}")
return self
def __exit__(self, exc_type, exc_val, exc_tb):
self.release()
# 使用示例
def deduct_stock_with_lock(product_id, quantity):
lock = RedisLock(r, f"stock:{product_id}", expire_seconds=10)
try:
if lock.acquire(timeout=5):
product = db.query("SELECT stock FROM products WHERE id = %s", (product_id,))
if product['stock'] < quantity:
return False
db.execute(
"UPDATE products SET stock = stock - %s WHERE id = %s",
(quantity, product_id)
)
return True
else:
raise RuntimeError("获取锁超时")
finally:
lock.release()
可重入锁实现
class ReentrantRedisLock:
def __init__(self, redis_client, key, expire_seconds=10):
self.redis = redis_client
self.key = f"lock:{key}"
self.expire_seconds = expire_seconds
self.client_id = str(uuid.uuid4())
self.hold_count_key = f"lock:hold:{self.key}:{self.client_id}"
def acquire(self, blocking=True, timeout=10):
-- 检查是否是同一客户端重入
hold_count = self.redis.get(self.hold_count_key)
if hold_count:
self.redis.incr(self.hold_count_key)
self.redis.expire(self.hold_count_key, self.expire_seconds)
return True
if not blocking:
return bool(self.redis.set(self.key, self.client_id, nx=True, ex=self.expire_seconds))
start_time = time.time()
while time.time() - start_time < timeout:
if self.redis.set(self.key, self.client_id, nx=True, ex=self.expire_seconds):
self.redis.setex(self.hold_count_key, self.expire_seconds, 1)
return True
time.sleep(0.01)
return False
def release(self):
hold_count = self.redis.get(self.hold_count_key)
if hold_count:
hold_count = int(hold_count)
if hold_count > 1:
self.redis.decr(self.hold_count_key)
else:
lua_script = """
if redis.call("get", KEYS[1]) == ARGV[1] then
return redis.call("del", KEYS[1])
else
return 0
end
"""
self.redis.eval(lua_script, 1, self.key, self.client_id)
self.redis.delete(self.hold_count_key)
Redlock 算法
单节点 Redis 锁在主节点故障时，如果从节点尚未同步最新的锁数据，可能导致锁丢失。Redlock 算法通过在多个独立 Redis 节点上获取锁来提高可靠性。
Redlock 原理：
1. 客户端向 N 个独立 Redis 节点并行发送获取锁请求
2. 如果超过 N/2+1 个节点成功获取锁，且总耗时小于锁过期时间，则认为成功获取锁
3. 实际锁的有效期 = 初始过期时间 - 总耗时
4. 释放锁时，向所有节点发送释放请求
class Redlock:
def __init__(self, redis_clients):
self.clients = redis_clients
self.quorum = len(redis_clients) // 2 + 1
self.client_id = str(uuid.uuid4())
def acquire(self, resource, ttl_ms=10000, retry_count=3):
ttl = ttl_ms / 1000
for _ in range(retry_count):
acquired_count = 0
start_time = time.time()
for client in self.clients:
if self._try_acquire(client, resource, ttl):
acquired_count += 1
elapsed = time.time() - start_time
validity_time = ttl - elapsed
if acquired_count >= self.quorum and validity_time > 0:
self._validity_time = validity_time
self._resource = resource
return True
self._release_all(resource)
time.sleep(0.05)
return False
def _try_acquire(self, client, resource, ttl):
lock_key = f"lock:{resource}"
return bool(client.set(lock_key, self.client_id, nx=True, ex=int(ttl)))
def _release_all(self, resource):
lock_key = f"lock:{resource}"
lua_script = """
if redis.call("get", KEYS[1]) == ARGV[1] then
return redis.call("del", KEYS[1])
else
return 0
end
"""
for client in self.clients:
try:
client.eval(lua_script, 1, lock_key, self.client_id)
except Exception:
pass
def release(self):
if hasattr(self, '_resource'):
self._release_all(self._resource)
分布式锁的常见错误
-- 错误1：使用 WATCH + MULTI/EXEC（不适合分布式锁）
-- WATCH 是乐观锁，只适合单实例场景
-- 错误2：锁的过期时间设置不当
-- 过期时间太短：业务还没执行完，锁就自动释放了
-- 过期时间太长：客户端崩溃后，锁要等很久才能自动释放
-- 错误3：单机 Redis 锁没有考虑主从故障
-- 如果使用 Redis 主从复制，主节点崩溃后，从节点晋升
-- 但锁信息可能还没同步到从节点，导致锁丢失
-- 错误4：没有考虑锁续期（Watch Dog 机制）
-- 如果业务执行时间超过锁过期时间，锁会被自动释放
-- 但业务还在执行，其他客户端获取了锁，导致数据竞争
常见问题
Q1：Redlock 真的安全吗？
Redlock 的作者 Martin Kleppmann 对 Redlock 提出了质疑，认为它在时钟跳跃、网络分区等极端场景下可能不安全。在大多数业务场景下，单节点 Redis 锁 + 足够长的过期时间已经足够，Redlock 的复杂度可能得不偿失。
Q2：锁的过期时间怎么定？
根据业务操作的最长耗时 * 2 作为初始过期时间。例如正常操作耗时 200ms，但可能有网络抖动或数据库慢查询，设为 5-10 秒比较安全。如果业务执行时间不确定，应使用 Watch Dog 自动续期。
Q3：分布式锁和数据库唯一索引冲突吗？
两者可以互补。数据库唯一索引是数据层面的保证，分布式锁是代码层面的保证。分布式锁的粒度更细，但可能因代码 bug 导致死锁；唯一索引更简单，但只能用于防止重复数据。
延伸阅读
- Redis 缓存策略与实战
- Redis 集群配置
- Redis 官方文档：Redis Distributed Locks https://redis.io/docs/manual/patterns/distributed-locks/