python - 如何在 Linux 上的 Python 中检查文件锁？

Question

我想要一个模块，它具有锁定文件、解锁文件、检查文件是否被锁定的功能，并且还具有一些使用此锁定功能来保存和读取 JSON 文件的功能。

在通用模块函数中检查文件是否被锁定的可行方法是什么？我想到了类似 locking_module.read_lock_status("file.csv") 的内容。下面是我尝试的模块代码和两个可以同时运行的脚本，一个写入文件，另一个从同一文件读取。

模块

import fcntl
import json
import time
import struct
import re

def lock(filepath):
    lock_file = open(filepath, "a")
    try:
        fcntl.lockf(lock_file, fcntl.LOCK_EX | fcntl.LOCK_NB)
    except (OSError, BlockingIOError) as error:
        return error
    return lock_file

def unlock(filepath):
    lock_file = open(filepath, "a")
    try:
        fcntl.lockf(lock_file, fcntl.LOCK_UN)
    except (OSError, BlockingIOError) as error:
        return error
    return lock_file

def read_lock(filepath):
    lock_file = open(filepath, "r")
    lock_data = struct.pack("hhllhh", fcntl.F_WRLCK, 0, 0, 0, 0, 0)
    try:
        lock_query = fcntl.fcntl(lock_file, fcntl.F_GETLK, lock_data)
        lock_status = struct.unpack("hhllhh", lock_query)[0]
    except OSError:
        return False
    return lock_status

def save_JSON(filepath, dictionary, hang_until_unlocked = True):
    if lock(filepath):
        with open(filepath, "w") as file_JSON:
            json.dump(dictionary, file_JSON)
        unlock(filepath)
        return True
    elif hang_until_unlocked:
        while not lock(filepath):
            time.sleep(0.1)
        with open(filepath, "w") as file_JSON:
            json.dump(dictionary, file_JSON)
        unlock(filepath)
        return True
    else:
        return False

def load_JSON(filepath, hang_until_unlocked = True):
    if lock(filepath):
        try:
            with open(filepath) as file_JSON:
                dictionary = json.load(file_JSON)
            unlock(filepath)
            return dictionary
        except:
            return False
    elif hang_until_unlocked:
        while not lock(filepath):
            time.sleep(0.1)
        try:
            with open(filepath) as file_JSON:
                dictionary = json.load(file_JSON)
            unlock(filepath)
            return dictionary
        except:
            return False
    else:
        return False

脚本1

import random

import lock

while True:
    config = {"a": 1, "b": random.randint(1, 2)}
    lock.save_JSON("config.json", config)

脚本2

import lock

while True:
    config = lock.load_JSON("config.json")
    if config:
        print(config)

Answer 1

您可以像这样使用子进程从命令行调用 lsof

import subprocess
file_lock = str(subprocess.check_output("lsof | grep file_name", shell=True))