跨两个独立进程锁定JSON文件:两种实现方法对比探讨
两种跨进程文件锁方法的优缺点分析
方法1:自定义文件锁(I/O锁)
实现代码
工具函数:
from pathlib import Path def acquire_file_lock(filename: str) -> None: LOCK_FILE_PATH = filename + ".lock" with open(LOCK_FILE_PATH, 'w') as _: pass def file_lock_exists(filename: str) -> bool: LOCK_FILE_PATH = filename + ".lock" return Path(LOCK_FILE_PATH).is_file() def remove_file_lock(filename: str) -> None: LOCK_FILE_PATH = filename + ".lock" Path(LOCK_FILE_PATH).unlink(missing_ok=True)
a.py中的调用代码:
import time def funcA(): while file_lock_exists("r.json"): time.sleep(0.1) try: acquire_file_lock("r.json") do_something_A() remove_file_lock("r.json") except: remove_file_lock("r.json")
b.py逻辑类似,仅替换funcA为funcB、do_something_A为do_something_B。
优点
- 实现逻辑直观易懂,基于基础文件操作,无需依赖系统特定扩展库
- 锁文件可见,可直接通过文件系统查看锁状态,便于问题排查
缺点
- 存在致命竞态条件:从检查锁文件不存在到创建锁文件的时间窗口内,多个进程可能同时通过检查并创建锁,导致锁失效
- 轮询机制低效:需要定时sleep轮询锁状态,sleep时长难以平衡——过长会导致响应延迟,过短则额外消耗CPU资源
- 锁残留风险:进程意外崩溃时,锁文件无法自动删除,会导致后续进程永久无法获取锁,需额外添加清理逻辑
- 无原子性保障:锁的检查、创建、删除是分离操作,无法保证原子性,这是竞态条件的根源
方法2:fcntl.flock内核级文件锁
实现代码
工具类:
import fcntl class FileLocker: def __init__(self, filename: str): self.filename_with_lock = filename def __enter__(self): self.fp = open(self.filename_with_lock) fcntl.flock(self.fp.fileno(), fcntl.LOCK_EX) def __exit__(self, _type, value, tb): fcntl.flock(self.fp.fileno(), fcntl.LOCK_UN) self.fp.close()
a.py中的调用代码:
def funcA(): with FileLocker("r.json"): try: do_something_A() except: pass
b.py逻辑类似,仅替换funcA为funcB、do_something_A为do_something_B。
优点
- 原子性锁操作:由操作系统内核处理锁的获取与释放,完全避免竞态条件,可靠性拉满
- 高效阻塞等待:锁被占用时进程会被内核挂起,不消耗CPU资源,锁释放后自动唤醒,无轮询的资源浪费与延迟问题
- 自动释放锁:进程正常退出或意外崩溃时,内核会自动回收文件描述符并释放锁,不会出现锁残留
- 代码简洁安全:通过上下文管理器(
with语句)自动管理锁的生命周期,减少手动操作出错的概率
缺点
- 平台局限性:仅适配POSIX系统(如Linux、macOS),无法在Windows环境使用(但你明确仅在Linux运行,此问题可忽略)
- 锁状态不直观:锁由内核维护,无法通过文件系统直接查看,排查问题需借助
lsof等系统工具 - 需注意文件描述符生命周期:若提前关闭锁对应的文件描述符会导致锁提前释放,但通过上下文管理器的写法可规避此问题
总结
如果你的场景仅限定在Linux环境,方法2是生产环境的首选方案,它彻底解决了方法1的核心缺陷,在可靠性、效率、易用性上都远超自定义文件锁。方法1仅适合低并发、非核心场景临时使用,或作为跨进程同步的学习示例。
内容的提问来源于stack exchange,提问作者Richard
相关产品推荐
相关产品推荐

