业务场景描述：
- 多个业务节点（多台主机运行同一个业务代码），但是只有一个节点真正运行（运行节点），其他都处于等待（备选节点）
- 运行节点一旦异常（服务挂掉，宕机，误操作等），其他备选节点，立刻有一个继续运行
- 之前的运行节点，重启之后，变为备选节点，处于等待节点
- 实现多节点高可用
分布式锁具备的功能：
- 资源唯一表示
- 锁续时（锁绝对不能不设置过期时间）
- 锁误操作（改变了不属于自己的锁）-----资源唯一表示，即可解决
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，可见参考
评论 (0)