Python多线程共享Config_File对象的线程安全与锁使用问题

初始版本读不到最新配置值 #

核心原因通常有三类：

• 读写逻辑绕开了内存实例：写操作直接写磁盘、读操作直接读磁盘文件，受操作系统文件缓存、Python IO缓存影响，无法拿到刚写入的最新值
• Flask运行模式不匹配：开了debug重载、多worker部署时，各个工作逻辑运行在不同进程中，每个进程持有独立的Config_File实例，内存完全隔离，自然无法读到其他进程写入的值
• 无锁读写导致的可见性问题：多个线程同时读写实例属性，夹杂IO操作时会出现指令重排，线程可能长时间读到属性的旧缓存值

加锁后无法启动多线程 #

锁选型完全错误：multiprocessing.Lock是为多进程跨进程互斥设计的，底层依赖进程间通信机制实现，在同进程多线程场景下使用会出现锁状态异常、死锁、线程调度阻塞，直接导致只有一个线程能正常运行。多线程场景的互斥锁必须使用threading模块提供的锁实现。

实现原则 #

• 单例模式保证全进程所有线程拿到的是同一个Config_File实例
• 所有配置读写操作统一加可重入互斥锁，禁止无锁直接访问内部配置属性
• 内存中的配置作为唯一实时可信源，磁盘仅做持久化用途，读操作直接读内存，避免IO缓存导致的延迟
• 写磁盘采用原子替换逻辑，避免写文件中途程序崩溃导致配置文件损坏

可直接复用的代码实现 #

Config_File线程安全类 #

import threading
import json
import os
from typing import Any

class Config_File:
    _instance = None
    _init_lock = threading.Lock()
    _rw_lock = threading.RLock()  # 可重入锁，支持同线程嵌套调用不出现死锁

    def __new__(cls, *args, **kwargs):
        # 双检锁实现单例，保证多线程场景下只初始化一个实例
        if not cls._instance:
            with cls._init_lock:
                if not cls._instance:
                    cls._instance = super().__new__(cls)
        return cls._instance

    def __init__(self, config_path: str = "config.json"):
        if hasattr(self, "_initialized"):
            return
        self.config_path = config_path
        self._config_data = {}
        self._initialized = True
        # 初始化时仅从磁盘加载一次配置到内存
        with self._rw_lock:
            self._load_from_disk_unsafe()

    def _load_from_disk_unsafe(self) -> None:
        """无锁内部方法，调用方必须提前持有_rw_lock"""
        if not os.path.exists(self.config_path):
            self._config_data = {}
            return
        try:
            with open(self.config_path, "r", encoding="utf-8") as f:
                self._config_data = json.load(f)
        except (json.JSONDecodeError, IOError):
            self._config_data = {}

    def _save_to_disk_unsafe(self) -> None:
        """无锁内部方法，调用方必须提前持有_rw_lock，原子写避免文件损坏"""
        tmp_path = f"{self.config_path}.tmp"
        with open(tmp_path, "w", encoding="utf-8") as f:
            json.dump(self._config_data, f, indent=2, ensure_ascii=False)
        os.replace(tmp_path, self.config_path)

    def get(self, key: str, default: Any = None) -> Any:
        """线程安全读配置，直接返回内存中的最新值"""
        with self._rw_lock:
            return self._config_data.get(key, default)

    def set(self, key: str, value: Any, persist: bool = True) -> None:
        """
        线程安全写配置
        :param persist: 是否写入后立刻落盘，批量修改场景可传False，最后统一调用save()
        """
        with self._rw_lock:
            self._config_data[key] = value
            if persist:
                self._save_to_disk_unsafe()

    def get_all(self) -> dict:
        """返回全量配置的副本，避免外部拿到内部引用绕过锁修改"""
        with self._rw_lock:
            return self._config_data.copy()

    def save(self) -> None:
        """手动触发全量配置落盘，用于批量修改后持久化、定时备份场景"""
        with self._rw_lock:
            self._save_to_disk_unsafe()

三类工作逻辑的调用示例 #

from flask import Flask, jsonify, request
from gpiozero import Button
import threading
import time

app = Flask(__name__)
# 全局唯一配置实例
config = Config_File()

# Flask Web服务逻辑
@app.route("/api/config", methods=["GET"])
def get_config():
    # 直接读内存，无IO延迟
    return jsonify(config.get_all())

@app.route("/api/config/<key>", methods=["POST"])
def update_config(key):
    val = request.json.get("value")
    config.set(key, val)
    return jsonify({"code": 0, "msg": "success"})

# gpiozero硬件IO监控逻辑
def btn_press_callback():
    # 硬件事件触发的配置修改，直接调用线程安全方法
    current_io_state = config.get("io_active", False)
    config.set("io_active", not current_io_state)

# 绑定硬件引脚回调
hw_btn = Button(17)
hw_btn.when_pressed = btn_press_callback

# 配置定时落盘线程逻辑
def periodic_persist():
    while True:
        time.sleep(30)
        config.save()

if __name__ == "__main__":
    # 启动守护式落盘线程，主进程退出时自动终止
    persist_t = threading.Thread(target=periodic_persist, daemon=True)
    persist_t.start()
    # Flask启动注意：关闭debug模式、关闭reloader，使用默认单进程多线程模式，保证所有逻辑共享同一个配置实例
    app.run(host="0.0.0.0", port=5000, debug=False, threaded=True, use_reloader=False)

• 所有配置读写必须通过get/set/get_all/save方法，禁止直接访问_config_data内部属性，否则锁机制完全失效
• 该方案仅适用于单进程多线程场景，如果需要多进程部署Flask（比如gunicorn多worker），内存存储配置的方案不再适用，需要替换为Redis等跨进程共享存储，或改用文件锁+原子写的纯文件方案
• 前端轮询接口不要加多余的缓存头，接口直接返回内存中的配置值即可保证实时性，不需要额外加延迟
• 多线程场景不要混用multiprocessing模块的锁、队列等原语，这类原语为跨进程设计，在线程场景使用会出现不可预期的阻塞问题

内容的提问来源于stack exchange，提问作者otuga