Python 功能点实现:数据热更新
文章在这里: https://www.jianshu.com/p/2d31f1c7ef63
代码在这里: https://github.com/simoncos/practical-python/tree/master/features/hot-reloading
关键词:热更新 | 热重载 | 定时更新 | 即时更新 | 缓存 | functools | cachetools | LRU | TTL
发到这里主要也想问问有没有更好的方法来做这件事(比如是不是要怎样搞个调度 / 消息系统,或者自省?)
Python 功能点实现:数据热更新
开一个线程专门做更新。更新线程里面是实时还是定时都可以自己控制。
我理解你想在Python里实现数据热更新,也就是在不重启程序的情况下更新内存中的数据。这个需求在配置管理、特征工程等场景很常见。
最直接的方式是用一个全局变量配合文件监听。下面这个例子展示了如何监控JSON配置文件的变化并自动更新内存数据:
import json
import time
import threading
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler
import os
class ConfigManager:
def __init__(self, config_path):
self.config_path = config_path
self.config_data = {}
self.load_config()
self.setup_watcher()
def load_config(self):
"""加载配置文件到内存"""
try:
with open(self.config_path, 'r', encoding='utf-8') as f:
self.config_data = json.load(f)
print(f"[{time.ctime()}] 配置已加载: {self.config_data}")
except Exception as e:
print(f"加载配置失败: {e}")
def setup_watcher(self):
"""设置文件监听器"""
event_handler = FileChangeHandler(self)
observer = Observer()
observer.schedule(event_handler, os.path.dirname(self.config_path), recursive=False)
observer.start()
# 后台运行监听线程
self.observer_thread = threading.Thread(target=self._keep_alive)
self.observer_thread.daemon = True
self.observer_thread.start()
def _keep_alive(self):
"""保持监听线程运行"""
while True:
time.sleep(1)
def get(self, key, default=None):
"""获取配置值"""
return self.config_data.get(key, default)
class FileChangeHandler(FileSystemEventHandler):
def __init__(self, config_manager):
self.config_manager = config_manager
def on_modified(self, event):
if event.src_path == self.config_manager.config_path:
print(f"[{time.ctime()}] 检测到配置文件变化")
self.config_manager.load_config()
# 使用示例
if __name__ == "__main__":
# 创建配置文件
config_path = "app_config.json"
initial_config = {"feature_enabled": True, "max_users": 100, "timeout": 30}
with open(config_path, 'w') as f:
json.dump(initial_config, f)
# 初始化配置管理器
manager = ConfigManager(config_path)
# 模拟应用运行
print("程序运行中,尝试修改app_config.json文件...")
print(f"当前配置: {manager.config_data}")
try:
while True:
# 业务逻辑中使用配置
if manager.get("feature_enabled"):
print(f"处理用户,当前上限: {manager.get('max_users')}")
time.sleep(5)
except KeyboardInterrupt:
print("\n程序退出")
这个方案的核心要点:
- 用
watchdog库监听文件变化 - 配置文件修改时自动触发重载
- 通过类封装管理配置状态
- 使用守护线程保持监听
如果你需要更高级的功能,比如多进程共享数据,可以考虑用multiprocessing.Manager或者Redis这类外部存储。对于简单的键值对,也可以直接用importlib.reload重新导入模块。
简单说就是:监听文件变化然后重新加载数据。
zookeeper 配置中心
运算的线程如何 load 新数据?
啊算我没说,线程是共享内存的…但是并行下面就有点麻烦了是不?
有没有什么参考?
多进程使用共享内存呀。。
我司方案,每秒从 s3 上把服务配置拉下来。
另外学架构,解决方案不需要贴代码的。因为一段代码肯定解决不了,没啥意义。
需要热更新的配置写到 config.py
然后 reload 就行了啊
redis
贴代码是为了说清楚现在的做法咯
也是缓存类的做法,定时应该还是可以的,即时麻烦一些。
更新很好做,但是保证更新过程中服务的可用,更新过程的数据一致性问题感觉楼主说得不足
说一下我们现在使用的一些方案吧,主要分为配置更新和算法模型更新
配置更新:
1、zookeeper 配置中心,基于订阅的形式
2、统一的字典服务,每次服务使用之前请求或者轮询请求
模型更新:
1、看模型大小情况,如果模型不大的话,用双指针的形式,单独开一个线程用于模型更新,更新完成之后指针切换,指针切换是原子操作,没有安全问题
2、多进程服务,采用共享内存存储模型,由于模型过大,加上更新过程中这个模型可以忍受脏数据,所以就是直接往共享内存里写了。。。
楼主有造轮子精神值得点赞,这个做 demo 玩玩可以,生产环境不行,只能做一些简单的订阅更新功能,
对于你所说的 “更新 python 对象数据”,我觉得要首先本着 CAP 原则,再分下面三种情况来设置业界的解决方案:
( 1 )更新配置文件:更新数据量较小,能容忍一定的时延,但是需要保证高可用—— zookeeper ;
( 2 )更新内存数据:数据量大,需要在多个进程间进行切换,短时间内(周期更新)对服务性能要求较高 —— redis ;
( 3 )更新数据频繁(实时更新):拆分为生成者消费者模型,用消息队列来解耦进程间的耦合度,如 Kafka、rocketMQ 等等。
多进程或者分布式应用不是自然就搞一个集中式的配置服务吗?

