多Python脚本共用RecordManager实例解决并发写文件问题咨询

这个场景我碰到过好几次——多个独立脚本要共享同一个磁盘上的记录，核心痛点就是并发写入时的文件冲突和数据丢失。针对macOS High Sierra和Ubuntu 16.04这两个系统，我给你推荐几个从易到难的简单方案，你可以根据自己的需求选：

方案1：原生文件锁（零依赖，最直接） #

这两个系统都支持fcntl模块的文件锁机制，你只需要给RecordManager的读写操作加上排他锁，保证同一时间只有一个进程能修改文件。

修改你的RecordManager类，在读写文件时加锁：

import fcntl
import pickle

class RecordManager:
    def __init__(self, file_path):
        self.file_path = file_path
        # 初始化空记录文件（如果不存在）
        try:
            with open(self.file_path, 'rb'):
                pass
        except FileNotFoundError:
            with open(self.file_path, 'wb') as f:
                pickle.dump({}, f)  # 这里用字典举例，你可以换成自己的结构

    def _lock(self, file_obj):
        # 加排他锁，其他进程会阻塞直到锁释放
        fcntl.flock(file_obj, fcntl.LOCK_EX)

    def _unlock(self, file_obj):
        fcntl.flock(file_obj, fcntl.LOCK_UN)

    def update(self, key, value):
        # 用rb+模式打开，避免直接覆盖文件
        with open(self.file_path, 'rb+') as f:
            self._lock(f)
            # 读取现有记录
            records = pickle.load(f)
            # 执行更新
            records[key] = value
            # 回到文件开头写入新内容
            f.seek(0)
            pickle.dump(records, f)
            # 截断旧内容（如果新记录比旧的短）
            f.truncate()
            self._unlock(f)

注意点： #

• 一定要用rb+模式，不能用w模式，否则会直接清空文件，锁也起不到作用
• 写完后要truncate()，防止旧内容的残留
• 所有脚本都要使用这个加了锁的RecordManager类，不然锁机制会失效

方案2：用第三方库简化（更省心，少写代码） #

如果觉得自己写锁逻辑麻烦，可以用filelock这个跨平台库，它已经封装了macOS和Linux的文件锁实现，不用自己处理fcntl的细节。

首先安装库：

pip install filelock

然后修改RecordManager：

from filelock import FileLock
import pickle

class RecordManager:
    def __init__(self, file_path):
        self.file_path = file_path
        # 单独的锁文件，避免和记录文件混在一起
        self.lock = FileLock(f"{file_path}.lock")
        # 初始化记录文件
        try:
            with open(self.file_path, 'rb'):
                pass
        except FileNotFoundError:
            with open(self.file_path, 'wb') as f:
                pickle.dump({}, f)

    def update(self, key, value):
        # 自动加锁/解锁，with块结束后会释放锁
        with self.lock:
            with open(self.file_path, 'rb+') as f:
                records = pickle.load(f)
                records[key] = value
                f.seek(0)
                pickle.dump(records, f)
                f.truncate()

这个方案几乎不需要改动你原来的业务逻辑，只需要套一层锁的with语句，出错概率极低，推荐优先用这个。

方案3：守护进程统一处理（适合高并发/复杂场景） #

如果你的脚本数量很多，或者除了更新记录还有其他交互需求（比如实时查询记录），可以把RecordManager做成一个单独的守护进程，其他脚本通过本地Socket发送请求，由守护进程统一处理文件读写——这样完全避免了并发写入的问题。

第一步：写守护进程脚本（record_daemon.py） #

import socket
import pickle
import threading

class RecordManager:
    def __init__(self, file_path):
        self.file_path = file_path
        try:
            with open(self.file_path, 'rb') as f:
                self.records = pickle.load(f)
        except FileNotFoundError:
            self.records = {}

    def update(self, key, value):
        self.records[key] = value
        # 这里不需要锁，因为只有这个进程在写文件
        with open(self.file_path, 'wb') as f:
            pickle.dump(self.records, f)

def handle_client(conn, manager):
    # 接收客户端的更新请求
    data = conn.recv(1024)
    if data:
        key, value = pickle.loads(data)
        manager.update(key, value)
    conn.close()

def run_daemon(file_path, host='localhost', port=9999):
    manager = RecordManager(file_path)
    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    sock.bind((host, port))
    sock.listen(5)
    print(f"Record守护进程已启动，监听{host}:{port}")
    while True:
        conn, addr = sock.accept()
        # 用线程处理每个客户端请求
        threading.Thread(target=handle_client, args=(conn, manager)).start()

if __name__ == "__main__":
    run_daemon("records.pkl")

第二步：其他脚本作为客户端调用 #

import socket
import pickle

def update_record(key, value, host='localhost', port=9999):
    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    sock.connect((host, port))
    # 发送更新数据
    sock.send(pickle.dumps((key, value)))
    sock.close()

# 调用示例
update_record("脚本1的结果", 456)

这个方案的好处是逻辑清晰，所有写入都由一个进程处理，完全不会有冲突；缺点是需要额外运行一个守护进程，部署稍微麻烦一点，但如果你的场景比较复杂，这个方案更可靠。

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