Python 使用 Redis 实现分布式锁(转）

Posted 2020-12-11 Tracydzf

转自：https://woodenrobot.me/2020/03/09/redis-lock/

业界常用解决方案

• 基于 MySql 等数据库的唯一索引
• 基于 ZooKeeper 临时有序节点
• 基于 Redis 的 NX EX 参数

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

分布式锁的特点

• 互斥性。在任意时刻，只有一个客户端能持有锁
• 锁超时。即使一个客户端持有锁的期间崩溃而没有主动释放锁，也需要保证后续其他客户端能够加锁成功
• 加锁和解锁必须是同一个客户端，客户端自己不能把别人加的锁给释放了。

版本一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

加锁过程

1. 首先需要为锁生成一个唯一的标识，这里使用 uuid;
2. 然后使用 setnx 设置锁，如果该锁之前不存在其他客户端的锁则加锁成功，接着设置锁的过期时间防止发生死锁并返回锁的唯一标示；
3. 如果设置失败先判断一下锁名所在的锁是否有过期时间，因为 setnx 和 expire 两个命令执行不是原子性的，可能会出现加锁成功但是设置超时时间失败出现死锁。如果不存在就给锁重新设置过期时间，存在就不断循环直到加锁时间超时加锁失败。

解锁过程

1. 首先整个解锁操作需要在一个 Redis 的事务中进行；
2. 使用 watch 监听锁，防止解锁时出现删除其他人的锁；
3. 查询锁名所在的标识是否与本次解锁的标识相同；
4. 如果相同则在事务中删除这个锁，如果删除过程中锁自动失效过期又被其他客户端拿到，因为设置了 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 。

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] then
        return redis.call("del",KEYS[1])
    else
        return 0
    end
    """
    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 主节点挂了就会出现一些问题，比如主节点加锁后没有同步到从节点，从节点升为主节点，就会出现锁的丢失。