使用python和redis实现分布式锁

业务场景描述：

• 多个业务节点（多台主机运行同一个业务代码），但是只有一个节点真正运行（运行节点），其他都处于等待（备选节点）
• 运行节点一旦异常（服务挂掉，宕机，误操作等），其他备选节点，立刻有一个继续运行
• 之前的运行节点，重启之后，变为备选节点，处于等待节点
• 实现多节点高可用

分布式锁具备的功能：

• 资源唯一表示
• 锁续时（锁绝对不能不设置过期时间）
• 锁误操作（改变了不属于自己的锁）-----资源唯一表示，即可解决
• Redis 挂掉（或者哨兵模式下，主节点挂掉，哨兵发现和选举延迟，或者redis整个服务挂掉），本地服务处理。
  • redis连接挂掉之后重启，锁可能还在，也可能没有了。没有之后，可能其他的备选节点会加锁成功，成为运行节点
  • 也就是说：Redis重启之后，之前的运行节点说不一定不是运行节点了。
  • 非运行节点，需要把本地的服务都停掉，确保只有一个运行节点在运行主要业务逻辑
• 额外要求：锁具备高可用，易用，其他业务代码侵入小。所以使用修饰器。

实现步骤：

唯一资源表示：

• Redis中的key选取某一个服务名称，value选取主机（或本地IP）+ 主进程ID作为标示

代码：

def redis_distributed_lock(lock_name: str):
    """
    redis分布式锁
    :param lock_name: redis中锁的名称
    :return:
    """
    def redis_function(func):
        def wrapper(self, *args, **kwargs):
            """
            High-availability operating mechanism based on redis
            :return: None
            """
            redis_model = RedisModel()
            lock_value = socket.gethostname() + "_" + str(threading.get_ident())
            # 没有注册上,休息后再尝试注册
            while True:
                try:
                    result = redis_model.set_ex_nx(key=lock_name, value=lock_value, seconds=60)
                    # 执行主体方法
                    func(*args, **kwargs)
                    # 获取锁的过期时间
                    ttl = redis_model.ttl(key=lock_name)
                    time.sleep(ttl if ttl > 1 else 1)
                except Exception as e:
                    self.logger.warning(
                        "Run redis_distributed_lock failed, cause: {}, traceback: {}".format(e, traceback.format_exc()))
                    time.sleep(5)

        return wrapper
    return redis_function

锁的续时间：

• 锁续时（锁绝对不能不设置过期时间）。
• 锁续时是个持续的操作（while True），尽量不能影响主体业务逻辑操作。我这里的处理是：续时操作，创建一个额外的线程（watch_dog）或者额外线程执行主体业务逻辑函数。
• 续时需要判断一下，Redis中的锁，还是否为本运行节点锁具有的锁。是的话，才进行续时，防止锁误操作。
• 不是的话，本地的服务可能需要都停掉，主进程处于备用状态，防止多节点运行。
  • 1、要删除该运行节点下的所有创建的线程，并且不影响主进程继续运行，并处于备用状态，
  • 2、如果子线程创建了进程，等等，需要将该节点下的所有进程也删除掉

代码：

修饰器：redis_distributed_lock

#!/usr/bin/env python
# -*- coding: UTF-8 -*-
"""
=================================================
@Project -> File 
@IDE    ：PyCharm
@Author ：Desmond.zhan
@Date   ：2021/12/7 3:07 下午
@Desc   ：redis分布式锁，运行执行leader节点的服务器运行，其他的阻塞，以达到高可用
通过注解方式，灵活性更强
==================================================
"""
import ctypes
import inspect
import multiprocessing
import socket
import threading
import time
import traceback

# 自定义Redis Model.
from xxxxxx import RedisModel


def _async_raise(tid, exctype):
    """raises the exception, performs cleanup if needed"""
    tid = ctypes.c_long(tid)
    if not inspect.isclass(exctype):
        exctype = type(exctype)
    res = ctypes.pythonapi.PyThreadState_SetAsyncExc(tid, ctypes.py_object(exctype))
    if res == 0:
        raise ValueError("invalid thread id")
    elif res != 1:
        # """if it returns a number greater than one, you're in trouble,
        # and you should call it again with exc=NULL to revert the effect"""
        ctypes.pythonapi.PyThreadState_SetAsyncExc(tid, None)
        raise SystemError("PyThreadState_SetAsyncExc failed")


def _async_stop_children_process():
    """
    杀死主进程下所有的子进程
    :return:
    """
    children_process = multiprocessing.active_children()
    for process in children_process:
        process.terminate()


def redis_distributed_lock(lock_name: str):
    """
    redis分布式锁
    :param lock_name: redis中锁的名称
    :return:
    """

    def redis_function(func):
        def wrapper(self, *args, **kwargs):
            """
            High-availability operating mechanism based on redis
            :return: None
            """
            redis_model = RedisModel()
            lock_value = socket.gethostname() + "_" + str(threading.get_ident())
            # 没有注册上,休息后再尝试注册
            while True:
                try:
                    result = redis_model.set_ex_nx(key=lock_name, value=lock_value, seconds=60)
                    # 加锁成功后，判断是否该主机+线程用的这个锁,如果是，设置一个超时时间。否则仍旧等待加锁
                    self.logger.info("Redis set_nx result：{}".format(result))
                    self.logger.info(
                        "redis_model.get(lock_name) : {}, lock_value:{}".format(redis_model.get(lock_name), lock_value))

                    # 加锁成功
                    if result and redis_model.get(lock_name).decode(encoding="utf8") == lock_value:
                        # 获取到redis分布式锁，开启子线程，进行执行目标函数
                        t_func = threading.Thread(target=func, args=args, kwargs=kwargs)
                        t_func.start()
                        # 启动看门狗，lock续时
                        redis_model.lock_watchdog_timeout(key=lock_name, value=lock_value)
                        # 停止函数执行
                        _async_raise(tid=t_func.ident, exctype=SystemExit)
                        _async_stop_children_process()
                    # 获取锁的过期时间
                    ttl = redis_model.ttl(key=lock_name)
                    time.sleep(ttl if ttl > 1 else 1)
                except Exception as e:
                    self.logger.warning(
                        "Run redis_distributed_lock failed, cause: {}, traceback: {}".format(e, traceback.format_exc()))
                    time.sleep(5)

        return wrapper
    return redis_function

续时操作：watch_dog

    def lock_watchdog_timeout(self, key, value, seconds=30):
        self.logger.info("key:{} lock watch dog threading start work ".format(self.get_key(key)))
        try:
            self._lock_extend(key, value, seconds)
        except Exception as e:
            self.logger.warning("redis lock lock_watchdog_timeout error, error info :{}".format(e))

    def _lock_extend(self, key, value, lifetime):
        while True:
            try:
                self.logger.debug("watch dog extend key time, key:{}".format(self.get_key(key)))
                # 判断redis中还是否为本进程持有的锁
                redis_value = self.redis_client.get(name=self.get_key(key))
                if redis_value != value:
                    self.logger.warning(
                        "Watch dog warning, redis lock value is not acquired, will suspend function execution")
                    self.logger.warning("this value:{}, redis value".format(value, redis_value))
                    # watch dog也没意义了，直接返回吧
                    return
                result = self.redis_client.expire(name=self.get_key(key), time=lifetime)
                self.logger.debug("{} extend result : {}".format(self.get_key(key), str(result)))
                time.sleep(lifetime // 2)
            except Exception as e:
                self.logger.warning(
                    "Run _lock_extend failed, cause: {}, traceback: {}".format(e, traceback.format_exc()))
                time.sleep(5)

另外：

• 如果除主进程之外，其他的某些子进程，备选节点也需要保留，可以在_async_stop_children_process()中根据进程名字，或者进程名字前缀，进行过滤，选择性删除子进程
• 如果是子线程创建子进程，子进程再创建进程，可以在_async_stop_children_process()方法中迭代，发现所有的进程，杀死进程之后，这些子进程下的子线程也就不存在了。
• 由于主进程可能创建子线程，而成为备选节点后，主进程仍旧存活，子线程需要清除掉，所有不能使用signal去处理子进程，这里是调用底层C语言暴露的API，可见参考

参考：

https://docs.python.org/zh-cn/3.7/library/ctypes.html#module-ctypes

https://docs.python.org/zh-cn/3.6/c-api/init.html?highlight=pythreadstate_setasyncexc

https://docs.python.org/3/library/multiprocessing.html

https://docs.python.org/zh-cn/3/library/signal.html?highlight=signal