如何在Python中可靠实现多函数同时写入JSON文件？

当游戏网站用户操作频繁时，并发执行JSON文件读写会出现数据结构损坏（比如未闭合的{}），核心原因是多个进程/线程同时读取、修改、写入文件，导致写入操作被中断或覆盖。以下是几种可靠的解决方案：

方案1：使用文件锁实现原子读写 #

通过文件锁确保同一时间只有一个进程能读写JSON文件，避免并发冲突。Python可以用fcntl模块（Linux/macOS）或msvcrt模块（Windows）实现跨平台锁：

import json
import fcntl
import os

def safe_update_user_data(user, number):
    file_path = 'example.json'
    with open(file_path, 'r+') as f:
        # 获取文件排他锁
        fcntl.flock(f, fcntl.LOCK_EX)
        # 读取数据
        data = json.load(f)
        # 修改数据
        data[user]['Data'] += number
        # 移动文件指针到开头并截断内容
        f.seek(0)
        f.truncate()
        # 写入数据
        json.dump(data, f, indent=4)
        # 锁会在文件关闭时自动释放

注意：Windows系统需要替换fcntl为msvcrt，调用msvcrt.locking(f.fileno(), msvcrt.LK_LOCK, os.path.getsize(file_path))实现锁机制

方案2：使用消息队列异步处理写入 #

把所有写入请求放入队列，由单个线程/进程统一处理写入操作，彻底避免并发冲突。可以用Python内置的queue模块配合线程：

import json
import queue
import threading
import time

# 创建全局写入队列
write_queue = queue.Queue()

def write_worker():
    """队列消费线程，负责统一执行JSON写入"""
    while True:
        if not write_queue.empty():
            user, number = write_queue.get()
            try:
                with open('example.json', 'r') as f:
                    data = json.load(f)
                data[user]['Data'] += number
                with open('example.json', 'w') as f:
                    json.dump(data, f, indent=4)
            except Exception as e:
                print(f"写入失败: {e}")
            finally:
                write_queue.task_done()
        time.sleep(0.1)

# 启动后台消费线程
threading.Thread(target=write_worker, daemon=True).start()

def update_user_data(user, number):
    """对外暴露的更新接口，将请求存入队列"""
    write_queue.put((user, number))

方案3：改用数据库存储（推荐） #

JSON文件天生不适合高并发写入场景，改用轻量数据库如SQLite（无需额外服务）或MySQL/PostgreSQL，利用数据库的事务和锁机制保证数据一致性：

以SQLite为例：

import sqlite3

# 初始化数据库（仅首次执行）
def init_user_db():
    conn = sqlite3.connect('game_users.db')
    cursor = conn.cursor()
    cursor.execute('''CREATE TABLE IF NOT EXISTS users
                      (username TEXT PRIMARY KEY, data INTEGER, nonce INTEGER)''')
    # 插入初始测试数据
    cursor.execute("INSERT OR IGNORE INTO users VALUES ('User', 999, 420)")
    conn.commit()
    conn.close()

def update_user_data(user, number):
    conn = sqlite3.connect('game_users.db')
    cursor = conn.cursor()
    # 使用原子更新语句，数据库自动处理并发锁
    cursor.execute("UPDATE users SET data = data + ? WHERE username = ?", (number, user))
    conn.commit()
    conn.close()

# 初始化数据库
init_user_db()

这种方案不仅解决并发问题，还支持更复杂的查询和数据操作，更适配游戏网站的长期业务发展。

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