在Python多进程或多线程并发操作文件的场景中，多个执行单元同时读写同一文件很容易导致数据错乱、内容覆盖等问题，文件独占锁可以限制同一时间只有一个执行单元能操作目标文件，是解决这类问题的核心方案。由于不同操作系统的文件锁实现机制不同，Linux和macOS使用fcntl模块，Windows使用msvcrt模块，需要针对性处理才能实现跨平台兼容。
fcntl模块的文件锁实现（Linux/macOS）
fcntl是类Unix系统下的文件控制接口，提供了对文件描述符的各种操作，其中包含文件锁相关功能。文件锁分为共享锁和独占锁两种，独占锁会阻止其他进程获取该文件的任何锁，适合写操作场景。
核心函数说明
- fcntl.flock(fd, operation)：对文件描述符fd执行锁操作，operation是锁类型参数
- 锁类型参数：
fcntl.LOCK_EX表示独占锁，fcntl.LOCK_SH表示共享锁，fcntl.LOCK_UN表示释放锁，fcntl.LOCK_NB表示非阻塞模式
使用示例
以下代码演示了在Linux/macOS系统下使用fcntl实现文件独占锁的完整流程：
import fcntl
import time
def write_with_lock(file_path, content):
# 以读写模式打开文件，获取文件描述符
with open(file_path, 'a+') as f:
fd = f.fileno()
try:
# 获取独占锁，阻塞模式直到获取成功
fcntl.flock(fd, fcntl.LOCK_EX)
print(f"进程 {time.time()} 获取文件锁成功，开始写入")
# 执行文件写入操作
f.write(content + 'n')
f.flush()
time.sleep(2) # 模拟耗时操作
print(f"进程 {time.time()} 写入完成，准备释放锁")
finally:
# 释放文件锁
fcntl.flock(fd, fcntl.LOCK_UN)
if __name__ == '__main__':
write_with_lock('test.txt', 'fcntl lock test content')
msvcrt模块的文件锁实现（Windows）
msvcrt是Windows系统下的Python标准库，提供了对Windows运行时库的访问，其中包含文件锁相关接口，仅能在Windows系统下使用。
核心函数说明
- msvcrt.locking(fd, mode, nbytes)：对文件描述符fd加锁，mode是锁模式，nbytes是加锁的字节数
- 锁模式参数：
msvcrt.LK_LOCK表示阻塞加锁，msvcrt.LK_NBLCK表示非阻塞加锁，msvcrt.LK_RLCK表示阻塞加读锁，msvcrt.LK_UNLCK表示释放锁
使用示例
以下代码演示了Windows系统下使用msvcrt实现文件独占锁的流程：
import msvcrt
import time
def write_with_lock(file_path, content):
# 以二进制读写模式打开文件
with open(file_path, 'ab+') as f:
fd = f.fileno()
try:
# 阻塞模式获取独占锁，锁整个文件（从0位置开始，锁1字节即可代表整个文件）
msvcrt.locking(fd, msvcrt.LK_LOCK, 1)
print(f"进程 {time.time()} 获取文件锁成功，开始写入")
# 移动文件指针到末尾写入内容
f.seek(0, 2)
f.write((content + 'n').encode('utf-8'))
f.flush()
time.sleep(2) # 模拟耗时操作
print(f"进程 {time.time()} 写入完成，准备释放锁")
finally:
# 移动指针到加锁位置，释放锁
f.seek(0)
msvcrt.locking(fd, msvcrt.LK_UNLCK, 1)
if __name__ == '__main__':
write_with_lock('test.txt', 'msvcrt lock test content')
跨平台文件锁封装实现
为了在不同系统下都能使用统一的文件锁接口，可以通过判断当前系统类型，选择对应的锁实现模块，封装成通用的文件锁工具类。
封装思路
- 通过
sys.platform判断当前运行系统，startswith('win')则为Windows系统，否则为类Unix系统 - 统一对外提供acquire和release方法，内部根据系统选择fcntl或msvcrt的实现逻辑
- 支持上下文管理器协议，方便使用with语句管理锁生命周期
完整封装代码
import sys
import time
class FileLock:
def __init__(self, file_path, timeout=None):
self.file_path = file_path
self.timeout = timeout
self.file_obj = None
self.is_locked = False
# 判断当前系统类型
if sys.platform.startswith('win'):
self.lock_module = 'msvcrt'
else:
self.lock_module = 'fcntl'
def acquire(self):
# 打开文件获取文件描述符
self.file_obj = open(self.file_path, 'a+')
fd = self.file_obj.fileno()
start_time = time.time()
while True:
try:
if self.lock_module == 'fcntl':
import fcntl
# 非阻塞模式尝试获取锁，失败则抛出异常
fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
self.is_locked = True
return True
else:
import msvcrt
# 非阻塞模式尝试获取锁
msvcrt.locking(fd, msvcrt.LK_NBLCK, 1)
self.is_locked = True
return True
except (IOError, OSError):
# 如果设置了超时时间，判断是否超时
if self.timeout is not None and (time.time() - start_time) > self.timeout:
self.file_obj.close()
self.file_obj = None
return False
# 未超时则等待后重试
time.sleep(0.1)
def release(self):
if self.is_locked and self.file_obj:
fd = self.file_obj.fileno()
try:
if self.lock_module == 'fcntl':
import fcntl
fcntl.flock(fd, fcntl.LOCK_UN)
else:
import msvcrt
self.file_obj.seek(0)
msvcrt.locking(fd, msvcrt.LK_UNLCK, 1)
finally:
self.file_obj.close()
self.is_locked = False
self.file_obj = None
def __enter__(self):
self.acquire()
return self.file_obj
def __exit__(self, exc_type, exc_val, exc_tb):
self.release()
# 跨平台使用示例
if __name__ == '__main__':
lock = FileLock('cross_platform_test.txt', timeout=5)
with lock as f:
print(f"跨平台获取锁成功，进程ID：{time.time()}")
f.write('cross platform lock testn')
f.flush()
time.sleep(2)
print("跨平台写入完成，释放锁")
注意事项
- fcntl的文件锁是建议性锁，需要所有进程都遵守锁规则才会生效，强制性锁需要额外系统配置
- msvcrt的locking函数加锁的字节范围仅对当前进程生效，不同进程的锁范围需要保持一致才能正确互斥
- 文件锁会在文件描述符关闭时自动释放，因此使用with语句管理文件生命周期可以避免锁泄漏
- 跨平台封装时需要注意不同系统下文件打开模式的差异，Windows下二进制模式打开更稳定