好的，各位观众老爷们，欢迎来到今天的“Redis 锁事：从单机到 Redlock 的华丽转身”讲座！ 🎩✨
今天咱们不聊虚头巴脑的理论，直接来点实在的，手把手教你用 Redis 数据结构打造一把稳如泰山的分布式锁，顺便聊聊 Redlock 理论，避免大家掉进“看起来很美”的坑里。
一、锁的前世今生：从单机到分布式，锁的焦虑
话说，在单机时代，锁这种东西，就像是你家卧室门上的那把锁，保护你的隐私（数据）。Java 里的 synchronized，C++ 里的 mutex，都是干这个的。简单粗暴，好用！
但是！时代变了！现在是分布式时代！你的数据不再乖乖地待在一台服务器上，而是分散在成千上万台机器上。这时候，你家卧室门上的锁就不好使了，因为你可能有好几间卧室，每间卧室都放着一样重要的东西，你得给每间卧室都装上一把锁，而且还得保证这些锁是“同一个锁”，才能保证同一时刻只有一个进程能访问你的数据。
这就引出了分布式锁的需求。分布式锁，就像是银行金库的大门锁，必须保证绝对的安全，绝对的唯一。
二、Redis 数据结构：锁的基石
Redis 提供了几种数据结构，可以用来实现分布式锁，但最常用的，也是最靠谱的，就是 SETNX (SET if Not eXists) 命令。
- SETNX key value： 如果 key 不存在，就设置 key 的值为 value，并返回 1；如果 key 已经存在，就什么也不做，并返回 0。
这个命令的原子性，简直就是为锁量身定做的！想象一下，多个进程同时执行 SETNX，只有一个进程能成功，拿到锁。
三、单机 Redis 锁：简单但脆弱的美好
让我们先来写一个简单的单机 Redis 锁：
import redis
import time
import uuid
class RedisLock:
def __init__(self, redis_host, redis_port, lock_name, lock_timeout=10):
self.redis_client = redis.Redis(host=redis_host, port=redis_port)
self.lock_name = lock_name
self.lock_timeout = lock_timeout
self.lock_value = str(uuid.uuid4()) # 锁的唯一标识
def acquire(self):
"""获取锁"""
end_time = time.time() + self.lock_timeout
while time.time() < end_time:
if self.redis_client.setnx(self.lock_name, self.lock_value):
self.redis_client.expire(self.lock_name, self.lock_timeout)
return True
time.sleep(0.01) # 短暂休眠，避免 CPU 占用过高
return False
def release(self):
"""释放锁"""
if self.redis_client.get(self.lock_name) == self.lock_value:
self.redis_client.delete(self.lock_name)
return True
return False
# Example Usage
if __name__ == '__main__':
lock = RedisLock("localhost", 6379, "my_lock")
if lock.acquire():
print("Got the lock!")
time.sleep(5) # 模拟业务逻辑
lock.release()
print("Released the lock!")
else:
print("Failed to get the lock.")这段代码看起来很美好，对不对？ 进程A调用 acquire()，如果成功获取锁，就执行业务逻辑，然后调用 release() 释放锁。 进程B在进程A持有锁的时候调用 acquire()，会进入循环等待，直到超时。
但是！ 别高兴得太早！ 这段代码有几个致命的缺陷：
- 锁超时问题： 如果进程A在持有锁的过程中崩溃了，或者发生了网络问题，导致
release()没有被执行，那么这个锁就会一直存在，导致其他进程永远无法获取锁。虽然我们设置了过期时间，但是如果在过期时间前，进程A又恢复了，或者网络恢复了，进程A以为自己还持有锁，实际上锁已经过期了，其他进程也可能拿到了锁，导致数据不一致。 - 误删锁问题： 假设进程A获取了锁，锁的过期时间是 10 秒。过了 9 秒，进程A的业务逻辑还没有执行完，Redis 锁自动过期了。这时候，进程B获取了锁。然后，进程A执行完业务逻辑，调用
release()释放锁，但是它释放的是进程B的锁！ 这就导致了误删锁的问题。
四、Redlock 理论：多点开花，安全加倍
为了解决单机 Redis 锁的缺陷，Redis 的作者 antirez 提出了 Redlock 算法。 Redlock 算法的核心思想是：不要把鸡蛋放在一个篮子里！ 💡
Redlock 算法需要多个 Redis 实例（通常是 5 个），这些实例之间是独立的，没有主从关系。 当一个进程想要获取锁时，它需要执行以下步骤：
- 获取当前时间。
- 依次向 N 个 Redis 实例发送 SETNX 命令，并设置锁的过期时间。
- 如果进程在大部分 Redis 实例上（N/2 + 1）成功获取了锁，并且获取锁的总耗时小于锁的有效时间，那么就认为获取锁成功。
- 如果获取锁成功，那么锁的有效时间就是锁的原始有效时间减去获取锁的总耗时。
- 如果进程获取锁失败，那么需要向所有 Redis 实例发送 DEL 命令，释放锁。
释放锁的过程也很简单：向所有 Redis 实例发送 DEL 命令，删除锁。
用一张表格来总结一下：
| 步骤 | 描述 |
|---|---|
| 1 | 获取当前时间。 |
| 2 | 依次向 N 个 Redis 实例发送 SETNX 命令，并设置锁的过期时间。每个实例的过期时间应该相同。 |
| 3 | 如果进程在大部分 Redis 实例上（N/2 + 1）成功获取了锁，并且获取锁的总耗时小于锁的有效时间，那么就认为获取锁成功。 |
| 4 | 如果获取锁成功，那么锁的有效时间就是锁的原始有效时间减去获取锁的总耗时。例如，如果锁的原始有效时间是 10 秒，获取锁的总耗时是 1 秒，那么锁的实际有效时间就是 9 秒。 |
| 5 | 如果进程获取锁失败（例如，没有在大部分 Redis 实例上获取到锁，或者获取锁的总耗时大于锁的有效时间），那么需要向所有 Redis 实例发送 DEL 命令，释放锁。 |
| 6 | 释放锁：向所有 Redis 实例发送 DEL 命令，删除锁。即使进程只在部分 Redis 实例上获取了锁，也需要向所有实例发送 DEL 命令，以确保锁被完全释放。 |
Redlock 算法的优势：
- 容错性： 即使部分 Redis 实例宕机，只要大部分实例可用，锁仍然可以正常工作。
- 避免锁超时问题： 锁的有效时间是动态计算的，可以避免进程在持有锁的过程中，锁自动过期的问题。
- 避免误删锁问题： 每个锁都有一个唯一的标识，只有持有锁的进程才能释放锁。
五、Redlock 代码实现：理论到实践的完美融合
import redis
import time
import uuid
class Redlock:
def __init__(self, redis_nodes, lock_name, lock_timeout=10):
self.redis_nodes = redis_nodes # Redis 节点列表，例如：[{"host": "localhost", "port": 6379}, {"host": "localhost", "port": 6380}, ...]
self.lock_name = lock_name
self.lock_timeout = lock_timeout
self.lock_value = str(uuid.uuid4())
self.quorum = len(redis_nodes) // 2 + 1 # 大多数 Redis 实例
def acquire(self):
"""获取锁"""
start_time = time.time()
success_count = 0
for node in self.redis_nodes:
try:
redis_client = redis.Redis(host=node["host"], port=node["port"])
if redis_client.setnx(self.lock_name, self.lock_value):
redis_client.expire(self.lock_name, self.lock_timeout)
success_count += 1
except Exception as e:
print(f"Error connecting to Redis node: {node}, error: {e}")
pass # 忽略连接错误
# 判断是否获取锁成功
if success_count >= self.quorum and (time.time() - start_time) < self.lock_timeout:
validity_time = self.lock_timeout - (time.time() - start_time)
print(f"Got the lock! Validity time: {validity_time}")
return True
else:
# 释放锁
self.release()
print("Failed to get the lock.")
return False
def release(self):
"""释放锁"""
for node in self.redis_nodes:
try:
redis_client = redis.Redis(host=node["host"], port=node["port"])
if redis_client.get(self.lock_name) == self.lock_value:
redis_client.delete(self.lock_name)
except Exception as e:
print(f"Error connecting to Redis node: {node}, error: {e}")
pass # 忽略连接错误
# Example Usage
if __name__ == '__main__':
redis_nodes = [
{"host": "localhost", "port": 6379},
{"host": "localhost", "port": 6380},
{"host": "localhost", "port": 6381},
{"host": "localhost", "port": 6382},
{"host": "localhost", "port": 6383}
]
lock = Redlock(redis_nodes, "my_redlock")
if lock.acquire():
print("Got the Redlock!")
time.sleep(5) # 模拟业务逻辑
lock.release()
print("Released the Redlock!")
else:
print("Failed to get the Redlock.")这段代码实现了一个简单的 Redlock 算法。它连接到多个 Redis 实例，尝试获取锁，如果获取锁成功，就执行业务逻辑，然后释放锁。
六、Redlock 的局限性：银弹神话的破灭
Redlock 算法虽然很强大，但并不是银弹！ 它仍然存在一些局限性：
- 时钟漂移问题： Redlock 算法依赖于多个 Redis 实例的时钟同步。如果 Redis 实例之间的时钟漂移过大，可能会导致锁的有效时间计算错误，甚至导致锁失效。
- 网络分区问题： 在极端情况下，如果发生了网络分区，导致进程无法连接到大部分 Redis 实例，那么可能会导致多个进程同时获取锁。
- 复杂度： Redlock 的实现相对复杂，需要维护多个 Redis 连接，处理各种异常情况。
七、Redlock 的替代方案：百花齐放，各显神通
既然 Redlock 不是银弹，那么有没有其他的替代方案呢？ 当然有！
- ZooKeeper： ZooKeeper 是一个分布式协调服务，可以用来实现分布式锁。 ZooKeeper 的锁机制是基于 ZooKeeper 的顺序临时节点实现的。
- etcd： etcd 是一个分布式键值存储系统，也可以用来实现分布式锁。etcd 的锁机制是基于 etcd 的 Lease 机制实现的。
- Consul： Consul 是一个服务网格解决方案，也可以用来实现分布式锁。Consul 的锁机制是基于 Consul 的 Key/Value 存储实现的。
选择哪种方案，取决于你的具体需求。 如果你需要一个高可用、强一致性的锁，那么 ZooKeeper 或 etcd 可能是更好的选择。 如果你需要一个轻量级的、易于使用的锁，那么 Redis 或 Consul 可能是更好的选择。
八、总结：锁的艺术，永无止境
今天的讲座就到这里。 我们从单机锁讲到 Redlock，又从 Redlock 讲到其他的替代方案。 锁的艺术，永无止境。 希望大家能够根据自己的实际情况，选择合适的锁方案，保护好自己的数据。
记住，没有完美的锁，只有最适合你的锁！ 😉 祝大家锁事顺利！ 🎉