在 Python 中锁定文件

Question

我需要锁定一个文件以便用 Python 写入。 它将同时从多个 Python 进程访问。 我在网上找到了一些解决方案，但大多数都无法达到我的目的，因为它们通常仅基于 Unix 或基于 Windows。

Answer 1

我找到了一个简单且有效的（！）实现 来自灰白蟒蛇。

简单使用 os.open(..., O_EXCL) + os.close() 在 Windows 上不起作用。

Answer 2

您可能会发现 pylocker 非常有用。 它可用于锁定文件或一般锁定机制，并且可以同时从多个 Python 进程进行访问。

如果您只是想锁定文件，其工作原理如下：

import uuid
from pylocker import Locker

#  create a unique lock pass. This can be any string.
lpass = str(uuid.uuid1())

# create locker instance.
FL = Locker(filePath='myfile.txt', lockPass=lpass, mode='w')

# aquire the lock
with FL as r:
    # get the result
    acquired, code, fd  = r

    # check if aquired.
    if fd is not None:
        print fd
        fd.write("I have succesfuly aquired the lock !")

# no need to release anything or to close the file descriptor, 
# with statement takes care of that. let's print fd and verify that.
print fd

Answer 3

如果您只需要 Mac/POSIX，则无需外部软件包即可使用。

import sys
import stat
import os


filePath = "<PATH TO FILE>"
if sys.platform == 'darwin':
  flags = os.stat(filePath).st_flags
  if flags & ~stat.UF_IMMUTABLE:
    os.chflags(filePath, flags & stat.UF_IMMUTABLE)

如果你想解锁文件只需更改，

  if flags & stat.UF_IMMUTABLE:
    os.chflags(filePath, flags & ~stat.UF_IMMUTABLE)

Answer 4

截至 2024 年 6 月更新

现在似乎有许多强大的、跨平台的、积极维护的解决方案。 其他答案和评论中引用最多的一些是：

• filelock
• Portalocker
• oslo.concurrency （如果您需要更通用的多进程同步实用程序）

原始答案

好吧，所以我最终使用了我编写的代码 这里，在我的网站上 ^{链接已失效，请在 archive.org 上查看} (也可以在 GitHub 上找到）。 我可以按以下方式使用它：

from filelock import FileLock

with FileLock("myfile.txt.lock"):
    # work with the file as it is now locked
    print("Lock acquired.")

Answer 5

其他解决方案引用了大量外部代码库。 如果您想自己动手，这里有一些跨平台解决方案的代码，该解决方案在 Linux / DOS 系统上使用相应的文件锁定工具。

try:
    # Posix based file locking (Linux, Ubuntu, MacOS, etc.)
    #   Only allows locking on writable files, might cause
    #   strange results for reading.
    import fcntl, os
    def lock_file(f):
        if f.writable(): fcntl.lockf(f, fcntl.LOCK_EX)
    def unlock_file(f):
        if f.writable(): fcntl.lockf(f, fcntl.LOCK_UN)
except ModuleNotFoundError:
    # Windows file locking
    import msvcrt, os
    def file_size(f):
        return os.path.getsize( os.path.realpath(f.name) )
    def lock_file(f):
        msvcrt.locking(f.fileno(), msvcrt.LK_RLCK, file_size(f))
    def unlock_file(f):
        msvcrt.locking(f.fileno(), msvcrt.LK_UNLCK, file_size(f))


# Class for ensuring that all file operations are atomic, treat
# initialization like a standard call to 'open' that happens to be atomic.
# This file opener *must* be used in a "with" block.
class AtomicOpen:
    # Open the file with arguments provided by user. Then acquire
    # a lock on that file object (WARNING: Advisory locking).
    def __init__(self, path, *args, **kwargs):
        # Open the file and acquire a lock on the file before operating
        self.file = open(path,*args, **kwargs)
        # Lock the opened file
        lock_file(self.file)

    # Return the opened file object (knowing a lock has been obtained).
    def __enter__(self, *args, **kwargs): return self.file

    # Unlock the file and close the file object.
    def __exit__(self, exc_type=None, exc_value=None, traceback=None):        
        # Flush to make sure all buffered contents are written to file.
        self.file.flush()
        os.fsync(self.file.fileno())
        # Release the lock on the file.
        unlock_file(self.file)
        self.file.close()
        # Handle exceptions that may have come up during execution, by
        # default any exceptions are raised to the user.
        if (exc_type != None): return False
        else:                  return True        

现在，AtomicOpen 可以在 with 块中使用，其中通常使用 open 语句。

警告：

• 如果在 Windows 和 Python 上运行在调用 exit 之前崩溃，我不确定锁定行为是什么。
• 这里提供的锁定是建议性的，而不是绝对的。 所有潜在的竞争进程都必须使用“AtomicOpen”类。
• 截至（2020 年 11 月 9 日）此代码仅锁定 Posix 系统上的可写文件。 在发布之后和在此日期之前的某个时刻，在只读文件上使用 fcntl.lock 变得非法。

Answer 6

这里有一个跨平台的文件锁定模块： Portalocker

虽然正如 Kevin 所说，写入文件如果可能的话，您希望避免同时来自多个进程。

如果您可以将问题硬塞到数据库中，那么您可以使用 SQLite。 它支持并发访问并处理自己的锁定。

Answer 7

我一直在寻找几种解决方案来做到这一点，我的选择是
oslo.concurrency

它功能强大且文档相对完善。 它基于紧固件。

其他解决方案：

• Portalocker：需要 pywin32，这是一个 exe 安装，因此无法通过 pip
• 紧固件：记录不足
• lockfile：已弃用
• flufl.lock：POSIX 系统的 NFS 安全文件锁定。
• simpleflock ：上次更新 2013-07
• zc.lockfile ：最后更新于 2016 年 6 月（截至 2017 年 3 月）
• lock_file ：最后更新于 2007-10

Answer 8

我更喜欢 lockfile - 独立于平台的文件锁定

Answer 9

锁定是特定于平台和设备的，但一般来说，您有几个选择：

1. 使用flock()或等效方法（如果您的操作系统支持它）。 这是建议性锁定，除非您检查锁定，否则它会被忽略。
2. 使用锁定-复制-移动-解锁方法，复制文件，写入新数据，然后移动它（移动，而不是复制 - 移动是 Linux 中的原子操作 - 检查你的操作系统），然后检查锁定文件的存在。
3. 使用目录作为“锁”。 如果您要写入 NFS，则这是必需的，因为 NFS 不支持集群（）。
4. 还可以在进程之间使用共享内存，但我从未尝试过； 它是非常特定于操作系统的。

对于所有这些方法，您必须使用自旋锁（失败后重试）技术来获取和测试锁。 这确实为不同步留下了一个小窗口，但它通常足够小，不会成为主要问题。

如果您正在寻找跨平台的解决方案，那么您最好通过其他机制登录到另一个系统（其次是上面的 NFS 技术）。

请注意，sqlite 与普通文件一样受到 NFS 的相同约束，因此您无法写入网络共享上的 sqlite 数据库并免费获得同步。

Answer 10

下面是如何使用 filelock 库的示例，该库类似于 Evan Fosmark 的实现：

from filelock import FileLock

lockfile = r"c:\scr.txt"
lock = FileLock(lockfile + ".lock")
with lock:
    file = open(path, "w")
    file.write("123")
    file.close()

with lock: 块中的任何代码都是线程安全的，这意味着它将在另一个线程之前完成访问该文件。

Answer 11

在操作系统级别协调对单个文件的访问充满了您可能不想解决的各种问题。

最好的选择是有一个单独的进程来协调对该文件的读/写访问。

Answer 12

锁定文件通常是特定于平台的操作，因此您可能需要考虑在不同操作系统上运行的可能性。 例如：

import os

def my_lock(f):
    if os.name == "posix":
        # Unix or OS X specific locking here
    elif os.name == "nt":
        # Windows specific locking here
    else:
        print "Unknown operating system, lock unavailable"

Answer 13

我一直在处理这样的情况，我从同一目录/文件夹中运行同一程序的多个副本并记录错误。 我的方法是在打开日志文件之前将“锁定文件”写入光盘。 程序在继续之前检查“锁定文件”是否存在，如果“锁定文件”存在则等待轮到它。

这是代码：

def errlogger(error):

    while True:
        if not exists('errloglock'):
            lock = open('errloglock', 'w')
            if exists('errorlog'): log = open('errorlog', 'a')
            else: log = open('errorlog', 'w')
            log.write(str(datetime.utcnow())[0:-7] + ' ' + error + '\n')
            log.close()
            remove('errloglock')
            return
        else:
            check = stat('errloglock')
            if time() - check.st_ctime > 0.01: remove('errloglock')
            print('waiting my turn')

编辑---
在考虑了上面关于过时锁的一些评论之后，我编辑了代码以添加对“锁定文件”的过时性的检查。 在我的系统上对这个函数进行数千次迭代的计时，从之前：

lock = open('errloglock', 'w')

到之后：的平均时间为 0.002066... 秒，

remove('errloglock')

所以我想我将从这个数量的 5 倍开始，以指示陈旧情况并监控问题情况。

另外，当我在处理时间时，我意识到我有一些并不是真正必要的代码：

lock.close()

我在 open 语句之后立即添加了这些代码，因此我在此编辑中删除了它。

Answer 14

这对我有用：
不要占用大文件，分成几个小文件
您创建文件 Temp，删除文件 A，然后将文件 Temp 重命名为 A。

import os
import json

def Server():
    i = 0
    while i == 0:
        try:        
                with open(File_Temp, "w") as file:
                    json.dump(DATA, file, indent=2)
                if os.path.exists(File_A):
                    os.remove(File_A)
                os.rename(File_Temp, File_A)
                i = 1
        except OSError as e:
                print ("file locked: " ,str(e))
                time.sleep(1)
            
            
def Clients():
    i = 0
    while i == 0:
        try:
            if os.path.exists(File_A):
                with open(File_A,"r") as file:
                    DATA_Temp = file.read()
            DATA = json.loads(DATA_Temp)
            i = 1
        except OSError as e:
            print (str(e))
            time.sleep(1)

Answer 15

场景是这样的：
用户请求文件来做某事。 然后，如果用户再次发送相同的请求，它会通知用户在第一个请求完成之前第二个请求尚未完成。 这就是为什么我使用锁定机制来处理这个问题。

这是我的工作代码：

from lockfile import LockFile
lock = LockFile(lock_file_path)
status = ""
if not lock.is_locked():
    lock.acquire()
    status = lock.path + ' is locked.'
    print status
else:
    status = lock.path + " is already locked."
    print status

return status