Python 多进程操作文件时文件锁失效问题如何解决？

import multiprocessing
import json
import time
import fcntl
def init():
with open(‘list.txt’, ‘w’) as f:
f.write(json.dumps({‘sids’: {}}))
def list(data=None):
if data is None:
with open(‘list.txt’, ‘r’) as f:
fcntl.lockf(f.fileno(), fcntl.LOCK_EX)
print(f.read())
return json.loads(f.read())
else:
with open(‘list.txt’, ‘w’) as f:
fcntl.lockf(f.fileno(), fcntl.LOCK_EX)
f.write(json.dumps(data))
def generate(sid):
target = ‘http://www.baidu.com’
data = list()
data[‘sids’][sid] = target
print(len(data[‘sids’]))
list(data)
time.sleep(5)
if name == ‘main’:
init()
processes = []
for i in range(100):
p = multiprocessing.Process(target=generate, args=(i,))
processes.append§
p.start()
for p in processes:
    p.join()

几乎所有进程都报 json 解码错误

    raise ValueError("No JSON object could be decoded")
ValueError: No JSON object could be decoded

这是哪里出了问题？

gougou168 1楼

data[‘sids’][sid] = target
print(len(data[‘sids’]))

sid from ???

itying888 2楼

问题核心： 在Python多进程环境下，标准文件锁（如fcntl、os.lockf）可能因进程独立内存空间而失效。关键在于使用跨进程的同步原语。

解决方案： 使用multiprocessing模块提供的锁（如Lock、RLock），它们基于共享内存或信号量实现，能确保多进程间的互斥访问。

代码示例：

import multiprocessing
import time
import json

def worker_process(lock, file_path, process_id):
    """模拟写入文件的进程"""
    data = {"process_id": process_id, "timestamp": time.time()}
    
    with lock:  # 获取跨进程锁
        try:
            # 读取现有数据
            with open(file_path, 'r') as f:
                existing = json.load(f)
        except (FileNotFoundError, json.JSONDecodeError):
            existing = []
        
        # 追加新数据
        existing.append(data)
        
        # 写入文件
        with open(file_path, 'w') as f:
            json.dump(existing, f, indent=2)
        
        print(f"进程 {process_id} 写入完成")

if __name__ == '__main__':
    FILE_PATH = "shared_data.json"
    NUM_PROCESSES = 5
    
    # 初始化跨进程锁
    manager = multiprocessing.Manager()
    lock = manager.Lock()
    
    # 清空或初始化文件
    with open(FILE_PATH, 'w') as f:
        json.dump([], f)
    
    # 创建并启动进程
    processes = []
    for i in range(NUM_PROCESSES):
        p = multiprocessing.Process(
            target=worker_process,
            args=(lock, FILE_PATH, i)
        )
        processes.append(p)
        p.start()
    
    # 等待所有进程完成
    for p in processes:
        p.join()
    
    # 验证结果
    with open(FILE_PATH, 'r') as f:
        result = json.load(f)
    print(f"\n最终文件包含 {len(result)} 条记录")

关键点说明：

multiprocessing.Manager().Lock() 创建跨进程锁，替代threading.Lock
锁作用域需覆盖完整的文件读写操作
if __name__ == '__main__' 保护代码避免子进程重复执行

一句话建议： 用multiprocessing的锁替代系统文件锁即可解决跨进程同步问题。

phonegap100 3楼

试试 list 函数直接把 f 返回出去, 并且让外部一直持有这个对象
之前碰到过类似的问题. 写了个

def lock (filename) 这样的函数
一直不生效

应该是在函数内部打开文件, 文件对象在函数完成后就关闭了. 锁也释放了

sinazl 4楼

python def set_lock(block=False): lock_f = open(os.path.join(os.path.dirname(os.path.abspath(sys.argv[0])), '.{0}.lock'.format(os.path.splitext(os.path.basename(sys.argv[0]))[0])), 'w') if block: fcntl.flock(lock_f.fileno(), fcntl.LOCK_EX) else: fcntl.flock(lock_f.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB) return lock_f 

eggper 5楼

Python 3.6.2 (v3.6.2:5fd33b5, Jul 8 2017, 04:57:36) [MSC v.1900 64 bit (AMD64)] on win32
Type “help”, “copyright”, “credits” or “license” for more information.
>>> data = list()
>>> target = ‘http://www.baidu.com’
>>> data[‘sids’][0] = target
Traceback (most recent call last):
File “<stdin>”, line 1, in <module>
TypeError: list indices must be integers or slices, not str

直接报错了，真的能这么写吗