Python+Redis从零打造分布式锁实战示例

import redis
import time
def acquire_lock(redis_client, lock_key):
    identifier = str(time.time()) + str(id(threading.current_thread()))
    acquired = redis_client.setnx(lock_key, identifier)
    return acquired
def release_lock(redis_client, lock_key, identifier):
    pipe = redis_client.pipeline(True)
    while True:
        try:
            pipe.watch(lock_key)
            if pipe.get(lock_key) == identifier:
                pipe.multi()
                pipe.delete(lock_key)
                pipe.execute()
                return True
            pipe.unwatch()
            break
        except redis.exceptions.WatchError:
            pass
    return False
# 使用示例
r = redis.Redis(host='localhost', port=6379, db=0)
lock_key = 'mylock'
if acquire_lock(r, lock_key):
    print("Lock acquired")
    # 执行临界区代码...
    release_lock(r, lock_key, identifier)

def acquire_lock_with_timeout(redis_client, lock_key, expire_time):
    identifier = str(time.time()) + str(id(threading.current_thread()))
    while True:
        result = redis_client.set(lock_key, identifier, nx=True, ex=expire_time)
        if result:
            return identifier
        time.sleep(0.1)
# 使用示例不变，但在acquire_lock_with_timeout函数中增加了expire_time参数

RELEASE_LOCK_SCRIPT = """
if redis.call("get", KEYS[1]) == ARGV[1] then
    return redis.call("del", KEYS[1])
else
    return 0
end
"""
def release_lock LUA(redis_client, lock_key, identifier):
    release_script = redis_client.register_script(RELEASE_LOCK_SCRIPT)
    released = release_script(keys=[lock_key], args=[identifier])
    return bool(released)
# 使用示例不变

import redis
import time
class RedLock:
    def __init__(self, masters, ttl):
        self.masters = [redis.Redis(host=host, port=port, db=db) for host, port, db in masters]
        self.quorum = len(self.masters) // 2 + 1
        self.ttl = ttl
    def lock(self, resource_id):
        expiration = time.time() + self.ttl
        while time.time() < expiration:
            acquired_count = 0
            locked_instances = []
            # 尝试在每个Redis实例上获取锁
            for master in self.masters:
                if master.set(resource_id, 'locked', nx=True, px=self.ttl):
                    acquired_count += 1
                    locked_instances.append(master)
            # 如果获得超过半数以上的锁，则认为成功获取分布式锁
            if acquired_count > self.quorum:
                return locked_instances
            # 清除已获取但未达到法定数量的锁
            for master in locked_instances:
                master.delete(resource_id)
            # 等待一段时间后重试
            time.sleep(0.1)
        raise Exception("Could not acquire lock")
    def unlock(self, resource_id, masters_with_lock):
        for master in masters_with_lock:
            # 使用Lua脚本保证原子性地删除锁（与前面简单实现中的释放锁方法类似）
            RELEASE_LOCK_SCRIPT = """
            if redis.call("get", KEYS[1]) == "locked" then
                return redis.call("del", KEYS[1])
            else
                return 0
            end
            """
            release_script = master.register_script(RELEASE_LOCK_SCRIPT)
            release_script(keys=[resource_id])
# 示例用法：
masters = [('localhost', 6379, 0), ('localhost', 6380, 0)]  # 假设有两个Redis实例
redlock = RedLock(masters, 10000)  # 设置锁的有效期为10秒
resource_id = 'my_resource'
try:
    masters_with_lock = redlock.lock(resource_id)
    print(f"Acquired Redlock on resource {resource_id}")
    # 执行临界区代码...
finally:
    redlock.unlock(resource_id, masters_with_lock)