西北有高楼，上与浮云齐。 --- 西北望高楼
FLASK Python 小知识点 Web开发
本文主要通过Flask项目说明如何实现动态加载运行中项目的配置。
- 动态配置加载有什么作用呢？
正常来说，如果我们项目运行起来之后，已经完成了配置的加载。此时，我们如果想要变更一些配置，实现如降级开关切换、更改路径等操作，通常情况下，我们可以通过平滑重启服务来重新加载配置，但是在某些时候，我们不想重启服务或者因为某些原因不能重启，那么我们就需要一些动态覆盖加载配置的方式。
- 举个例子：
我们创建一个Flask应用，并加载项目配置。
# config.py
LOG_DIR = "/opt/logs/test_api"
LOG_LEVEL = INFO
# main.py
app = Flask(__name__)
app.config.from_object("config file name")
上面的代码展示了，我们通过一个配置文件加载App配置（比如我们操作日志记录）。当我们启动服务，记录的日志会存储到 /opt/logs/test_api 中。
当我们程序运行了一段时间之后，突然收到要求，需要我们把日志存放到指定目录 /opt/logs/logcenter 下，方便日志中心统一采集。
这个时候我们就可以尝试通过动态加载配置的方式去替换原来的日志路径。
Flask是支持动态加载配置的，我们在项目运行中，直接通过再执行 app.config.from_object 就可以重新加载配置，重新加载属于新增与覆盖式加载。
- Master-Worker启用多进程服务的动态加载
通过 Master-Worker 启用的服务，我们在动态加载配置的时候，需要做到每一个 Worker 进程都要更新，这样才能保证，我们在使用的时候，配置是正确的。
需要变更的时候，我们通过接口、MQ等方式，将配置更新到Redis或config文件或其他地方。
- 通过接口将配置更新到
Redis
# api.py
@check_api_bp.route('/conf/update', methods=['POST'])
def update_conf():
"""更新配置"""
log_dir = get_json('logDir', '配置', valid=[Required()], parm_type=str)
redis_store.set("LOG_DIR", log_dir)
return success(msg='更新成功')
- 方法一：使用时动态获取配置
我们可以实现一个装饰器，通过在每一个调用的函数上增加该装饰器。
# decorators.py
def _check_log_dir():
log_dir = redis_store.get("LOG_DIR")
if log_dir:
current_app.config["LOG_DIR"] = log_dir
current_app.log_dir = log_dir
def check_log_dir(func):
@wraps(func)
def wrapper_func(*args, **kwargs):
_check_switch()
return func(*args, **kwargs)
return wrapper_func
# api.py
@check_api_bp.route('/net/check', methods=['GET'])
@check_log_dir
def net_check():
"""服务健康检查"""
error_logger(current_app).info("服务健康检查")
return success(msg='请求成功')
上面伪代码说明：在调用 /net/check 接口的时候，需要用到日记记录，那么，我们就在该函数上，增加 check_log_dir 的装饰器。该装饰器会读取 Redis 里的 LOG_DIR 配置，如果读取到就覆盖加载到 app.config 上。在使用的时候，就能使用最新的配置了。
- 方法二：定时任务动态更新配置
定时任务动态更新和使用时更新差不多，只不过有一点要明确，在多进程服务中，需要在每一个进程下都执行定时任务，这样才能保证使用的 App 能正确的更新配置。
# crontab.py
# 此处使用的定时任务库 FLASK-APSCHEDULER
@scheduler_store.task("cron", id="crontab_host_update_config", minute="*/5")
def crontab_host_update_config():
current_app = scheduler_store.app
with current_app.app_context():
log_msg = "执行定时任务(minute=*/5)：{}".format(__name__)
log_dir = redis_store.get("LOG_DIR")
if log_dir:
current_app.config["LOG_DIR"] = log_dir
current_app.log_dir = log_dir
注意: 通过定时任务的方式去动态加载配置，有一定的时延。
上面就是动态加载配置的个人一点心得，可以一起探讨下。
网友评论