3. 规则集加载与热更新

任务目标

实现一个规则集加载器：从 JSON 文件读取规则集，并在文件被修改后自动重新加载，无需重启服务。核心是记录文件修改时间（mtime），在每次取规则时检查是否变化，变化则重新解析。

规则文件格式

规则集 JSON 结构：一个包含 rules 数组的对象，每条规则含 id、field、op、value。加载器只负责解析与热更新，不负责执行规则。

完整实现

import json
import os
import threading
import time

class RuleSetLoader:
    """从 JSON 文件加载规则集，文件修改后自动重新加载。"""

    def __init__(self, path):
        self.path = path
        self._lock = threading.Lock()
        self._mtime = None
        self._rules = []
        self._load(force=True)

    def _load(self, force=False):
        mtime = os.path.getmtime(self.path)
        if not force and mtime == self._mtime:
            return False
        with open(self.path, "r", encoding="utf-8") as f:
            data = json.load(f)
        rules = data.get("rules", [])
        for r in rules:
            if not all(k in r for k in ("id", "field", "op", "value")):
                raise ValueError(f"规则缺少必要字段: {r}")
        self._rules = rules
        self._mtime = mtime
        return True

    def get_rules(self):
        """每次调用都检查文件是否变化，变化则热更新。"""
        with self._lock:
            self._load()
            return list(self._rules)

    @property
    def mtime(self):
        return self._mtime

if __name__ == "__main__":
    path = "rules.json"
    if not os.path.exists(path):
        with open(path, "w", encoding="utf-8") as f:
            json.dump({"rules": [
                {"id": "R1", "field": "score", "op": ">=", "value": 700},
                {"id": "R2", "field": "age", "op": "<", "value": 60},
            ]}, f, ensure_ascii=False)

    loader = RuleSetLoader(path)
    print("初始规则:", loader.get_rules())

    # 模拟外部修改文件
    time.sleep(1.1)  # 确保 mtime 变化
    with open(path, "w", encoding="utf-8") as f:
        json.dump({"rules": [
            {"id": "R1", "field": "score", "op": ">=", "value": 750},
            {"id": "R3", "field": "income", "op": ">", "value": 5000},
        ]}, f, ensure_ascii=False)

    print("修改后规则:", loader.get_rules())

运行与验证

1. 保存代码为 loader.py，运行 python loader.py。
2. 首次运行会自动创建 rules.json，输出初始规则（R1、R2）。
3. 脚本内模拟修改文件后再次调用 get_rules()，输出应变为 R1（value=750）与 R3。
4. 判定通过：两次输出内容不同，且第二次反映文件最新内容，说明热更新生效。

关键点

• _load 通过比较 os.path.getmtime 判断是否需要重新解析，避免每次读盘。
• get_rules 每次调用都触发检查，因此外部修改文件后无需重启即可生效。
• 使用 threading.Lock 保证多线程下加载与读取不冲突。
• 校验规则必要字段，格式错误时抛出 ValueError，避免脏数据进入引擎。