Python JSON文件数据丢失求助：用户及积分莫名消失

问题概述 #

程序运行数小时后，JSON存储的用户数据（如user_registration.json）出现部分用户消失，对应积分文件也被重置。初始状态下排行榜能正常显示所有用户，但一段时间后出现异常。尝试过append模式保存文件，未解决问题。用户数据格式示例：

{"discordid1": {"username": "discorduser1", "roblox_username": "robloxuser1"}, "discordid2": {"username": "discorduser2", "roblox_username": "robloxuser2"}, "discordid3": {"username": "discorduser3", "roblox_username": "robloxuser3"}}

相关加载与保存代码：

import json
import os
import datetime as dt

COOLDOWNS_FILE = 'cooldowns.json'
SECOND_COOLDOWNS_FILE = 'second_cooldowns.json'
USER_LOGS_FILE = "user_logs.json"
USER_REGISTRATION_FILE = "user_registration.json"

def load_json_file(file_path):
    try:
        with open(file_path, 'r') as f:
            return json.load(f)
    except FileNotFoundError:
        return {}
    except json.JSONDecodeError:
        print(f"Error loading {file_path}: file is empty or invalid JSON")
        return {}

def save_json_file(data, file_path):
    with open(file_path, "w") as f:
        json.dump(data, f, indent=4)

def load_cooldowns(file_path):
    if not os.path.exists(file_path):
        return {}
    try:
        with open(file_path, "r") as f:
            data = json.load(f)
        data_to_return = {}
        for user_id, cooldown_time in data.items():
            data_to_return[user_id] = dt.datetime.fromisoformat(cooldown_time.replace("Z", "+00:00"))
        return data_to_return
    except json.JSONDecodeError:
        return {}

def save_cooldowns(cooldowns):
    data_to_save = {}
    for user_id, cooldown_time in cooldowns.items():
        data_to_save[user_id] = cooldown_time.isoformat()
    with open("cooldowns.json", "w") as f:
        json.dump(data_to_save, f, indent=4)

def load_second_cooldowns():
    if not os.path.exists(SECOND_COOLDOWNS_FILE):
        return {}
    try:
        with open(SECOND_COOLDOWNS_FILE, "r") as f:
            data = json.load(f)
        data_to_return = {}
        for user_id, cooldown_time in data.items():
            data_to_return[user_id] = dt.datetime.fromisoformat(cooldown_time)
        return data_to_return
    except json.JSONDecodeError:
        return {}

def save_second_cooldowns(second_cooldowns):
    data_to_save = {}
    for user_id, cooldown_time in second_cooldowns.items():
        data_to_save[user_id] = cooldown_time.isoformat()
    with open(SECOND_COOLDOWNS_FILE, "w") as f:
        json.dump(data_to_save, f, indent=4)

cooldowns = load_cooldowns(COOLDOWNS_FILE)
second_cooldowns = load_second_cooldowns()

问题原因分析 #

1. 并发读写冲突：如果程序是多线程/多进程架构，多个线程同时加载、修改、保存JSON文件时，会出现数据覆盖。比如线程A加载数据后，线程B也加载并先完成保存，线程A后续的保存会直接覆盖线程B的修改，导致部分数据丢失。
2. 非原子写入导致文件损坏：使用w模式打开文件时，会先清空原文件再写入。如果写入过程中程序崩溃、被中断，文件会变成空或不完整JSON，后续加载时触发JSONDecodeError，返回空字典，导致数据被重置。
3. 全局缓存数据过期：全局变量（如cooldowns）加载后未及时更新，修改时基于旧数据操作，保存后覆盖最新的文件内容。

解决方案 #

1. 加锁解决并发冲突 #

为每个JSON文件添加互斥锁，确保同一时间只有一个线程能读写文件：

import threading

# 为每个JSON文件创建独立锁
file_locks = {
    COOLDOWNS_FILE: threading.Lock(),
    SECOND_COOLDOWNS_FILE: threading.Lock(),
    USER_REGISTRATION_FILE: threading.Lock(),
    USER_LOGS_FILE: threading.Lock()
}

# 修改load_json_file函数，加入锁
def load_json_file(file_path):
    lock = file_locks.get(file_path, threading.Lock())
    with lock:
        try:
            with open(file_path, 'r') as f:
                return json.load(f)
        except FileNotFoundError:
            return {}
        except json.JSONDecodeError:
            print(f"Error loading {file_path}: file is empty or invalid JSON")
            return {}

# 修改save_json_file函数，加入锁和原子写入逻辑
def save_json_file(data, file_path):
    lock = file_locks.get(file_path, threading.Lock())
    with lock:
        temp_file = f"{file_path}.tmp"
        try:
            # 先写入临时文件
            with open(temp_file, "w") as f:
                json.dump(data, f, indent=4)
            # 原子替换原文件，避免写入中断损坏原文件
            os.replace(temp_file, file_path)
        except Exception as e:
            print(f"Error saving {file_path}: {str(e)}")
            # 清理临时文件
            if os.path.exists(temp_file):
                os.remove(temp_file)

2. 原子写入避免文件损坏 #

复用原子写入逻辑修改冷却时间保存函数：

def save_cooldowns(cooldowns):
    data_to_save = {}
    for user_id, cooldown_time in cooldowns.items():
        data_to_save[user_id] = cooldown_time.isoformat()
    save_json_file(data_to_save, COOLDOWNS_FILE)

def save_second_cooldowns(second_cooldowns):
    data_to_save = {}
    for user_id, cooldown_time in second_cooldowns.items():
        data_to_save[user_id] = cooldown_time.isoformat()
    save_json_file(data_to_save, SECOND_COOLDOWNS_FILE)

3. 避免全局缓存过期 #

修改全局变量更新逻辑，确保每次修改基于最新文件内容：

def update_cooldown(user_id, new_cooldown_time):
    global cooldowns
    # 加锁加载最新数据
    lock = file_locks[COOLDOWNS_FILE]
    with lock:
        cooldowns = load_cooldowns(COOLDOWNS_FILE)
        cooldowns[user_id] = new_cooldown_time
        save_cooldowns(cooldowns)

4. 增强异常日志记录 #

添加详细日志方便排查问题：

import logging

logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')

def load_json_file(file_path):
    lock = file_locks.get(file_path, threading.Lock())
    with lock:
        try:
            with open(file_path, 'r') as f:
                data = json.load(f)
                logging.info(f"Loaded {len(data)} entries from {file_path}")
                return data
        except FileNotFoundError:
            logging.info(f"{file_path} not found, returning empty dict")
            return {}
        except json.JSONDecodeError as e:
            logging.error(f"JSON decode error for {file_path}: {str(e)}")
            return {}

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