多进程环境下安全使用pickle实现持久化存储

需求背景 #

我在Python应用中需要一种非常简单的持久化存储方案。平时会用dict存数据，变更时用pickle/json写到磁盘，启动时读回来，但这种方式只适用于单进程场景。多进程环境下就算每次操作前读数据，还是会有竞态条件——比如某进程读不到另一进程的写入，或者多个进程同时写入导致数据损坏。

性能不是问题，写入操作极少（间隔几分钟到几小时），数据量也极小（只有几十条记录），但必须避免数据损坏。专业数据库太冗余，不想加额外组件；SQLite相对合适，但不确定怎么在多写入场景下安全使用。

之前好像见过Python里能实现这个功能的库，甚至可能是标准库组件——要么是基于磁盘的"魔法字典"，要么是能锁定文件、序列化访问的上下文管理器，类似这样的代码：

with FileLocker(filename, "r+b") as f:
    db = pickle.load(f)
    db["setting"] = "new value"
    pickle.dump(db, f)

可行解决方案 #

1. 标准库实现文件锁（跨平台） #

自己写一个轻量的上下文管理器，利用系统级文件锁实现序列化访问，完全基于标准库，无需额外依赖：

import os
import pickle
from contextlib import contextmanager

@contextmanager
def file_locker(filename, mode):
    # 初始化空字典文件（如果不存在）
    if not os.path.exists(filename):
        with open(filename, 'wb') as init_f:
            pickle.dump({}, init_f)
    
    fd = os.open(filename, os.O_RDWR)
    try:
        # Unix/Linux 独占锁
        if os.name == 'posix':
            import fcntl
            fcntl.flock(fd, fcntl.LOCK_EX)
        # Windows 文件锁
        elif os.name == 'nt':
            import msvcrt
            file_size = os.path.getsize(filename)
            msvcrt.locking(fd, msvcrt.LK_LOCK, file_size)
        
        # 转换为文件对象并交给上下文
        with os.fdopen(fd, mode) as f:
            yield f
    finally:
        # 解锁并关闭文件描述符
        if os.name == 'nt':
            import msvcrt
            file_size = os.path.getsize(filename)
            msvcrt.locking(fd, msvcrt.LK_UNLCK, file_size)
        os.close(fd)

# 使用示例
with file_locker('data.pkl', 'rb+') as f:
    db = pickle.load(f)
    db['setting'] = 'new value'
    f.seek(0)  # 回到文件开头
    pickle.dump(db, f)
    f.truncate()  # 截断旧数据，避免残留

注意：Windows下锁机制和Unix略有差异，必须确保操作完成后解锁；写入后执行truncate是为了防止新数据长度短于旧数据时，残留的旧数据导致反序列化错误。

2. 标准库SQLite（推荐） #

SQLite是Python标准库自带的轻量数据库，天生支持多进程并发安全，内置文件锁机制保证数据不损坏，完全符合你的需求：

import sqlite3

def init_db():
    conn = sqlite3.connect('settings.db')
    # 创建键值对表（首次运行自动创建）
    conn.execute('''CREATE TABLE IF NOT EXISTS settings
                    (key TEXT PRIMARY KEY, value TEXT)''')
    conn.commit()
    conn.close()

def set_setting(key, value):
    conn = sqlite3.connect('settings.db')
    try:
        # 存在则替换，不存在则插入
        conn.execute('REPLACE INTO settings (key, value) VALUES (?, ?)', (key, value))
        conn.commit()
    finally:
        conn.close()

def get_setting(key):
    conn = sqlite3.connect('settings.db')
    try:
        cursor = conn.execute('SELECT value FROM settings WHERE key = ?', (key,))
        result = cursor.fetchone()
        return result[0] if result else None
    finally:
        conn.close()

# 初始化（只需运行一次，也可以放在set/get里自动判断）
init_db()

# 使用示例
set_setting('setting', 'new value')
print(get_setting('setting'))

SQLite会自动处理并发写入的锁冲突，无需手动加锁，操作简单且绝对安全，是这种场景下的最优解之一。

3. 标准库shelve（结合锁） #

shelve是标准库提供的键值对存储，类似"磁盘字典"，但本身不支持多进程写安全，需要结合上面的文件锁使用：

import shelve

# 复用前面的file_locker上下文管理器
with file_locker('shelve_db', 'r+b') as f:
    with shelve.open(f.fileno()) as db:
        db['setting'] = 'new value'
        print(db['setting'])

这种方式可以实现类似"魔法字典"的用法，但需要额外加锁，不如SQLite省心。

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