Python 功能点实现:数据热更新

文章在这里: https://www.jianshu.com/p/2d31f1c7ef63

代码在这里: https://github.com/simoncos/practical-python/tree/master/features/hot-reloading

关键词:热更新 | 热重载 | 定时更新 | 即时更新 | 缓存 | functools | cachetools | LRU | TTL

发到这里主要也想问问有没有更好的方法来做这件事(比如是不是要怎样搞个调度 / 消息系统,或者自省?)


Python 功能点实现:数据热更新

15 回复

开一个线程专门做更新。更新线程里面是实时还是定时都可以自己控制。


我理解你想在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程序退出")

这个方案的核心要点:

  1. watchdog库监听文件变化
  2. 配置文件修改时自动触发重载
  3. 通过类封装管理配置状态
  4. 使用守护线程保持监听

如果你需要更高级的功能,比如多进程共享数据,可以考虑用multiprocessing.Manager或者Redis这类外部存储。对于简单的键值对,也可以直接用importlib.reload重新导入模块。

简单说就是:监听文件变化然后重新加载数据。

zookeeper 配置中心

运算的线程如何 load 新数据?

啊算我没说,线程是共享内存的…但是并行下面就有点麻烦了是不?

有没有什么参考?

多进程使用共享内存呀。。

我司方案,每秒从 s3 上把服务配置拉下来。
另外学架构,解决方案不需要贴代码的。因为一段代码肯定解决不了,没啥意义。

需要热更新的配置写到 config.py
然后 reload 就行了啊

贴代码是为了说清楚现在的做法咯

也是缓存类的做法,定时应该还是可以的,即时麻烦一些。

更新很好做,但是保证更新过程中服务的可用,更新过程的数据一致性问题感觉楼主说得不足

说一下我们现在使用的一些方案吧,主要分为配置更新和算法模型更新
配置更新:
1、zookeeper 配置中心,基于订阅的形式
2、统一的字典服务,每次服务使用之前请求或者轮询请求

模型更新:
1、看模型大小情况,如果模型不大的话,用双指针的形式,单独开一个线程用于模型更新,更新完成之后指针切换,指针切换是原子操作,没有安全问题
2、多进程服务,采用共享内存存储模型,由于模型过大,加上更新过程中这个模型可以忍受脏数据,所以就是直接往共享内存里写了。。。

楼主有造轮子精神值得点赞,这个做 demo 玩玩可以,生产环境不行,只能做一些简单的订阅更新功能,
对于你所说的 “更新 python 对象数据”,我觉得要首先本着 CAP 原则,再分下面三种情况来设置业界的解决方案:
( 1 )更新配置文件:更新数据量较小,能容忍一定的时延,但是需要保证高可用—— zookeeper ;
( 2 )更新内存数据:数据量大,需要在多个进程间进行切换,短时间内(周期更新)对服务性能要求较高 —— redis ;
( 3 )更新数据频繁(实时更新):拆分为生成者消费者模型,用消息队列来解耦进程间的耦合度,如 Kafka、rocketMQ 等等。

多进程或者分布式应用不是自然就搞一个集中式的配置服务吗?

回到顶部