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

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

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

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

其他解决方案引用了大量的外部代码库。如果您更愿意自己动手，这里有一些跨平台解决方案的代码，在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是非法的。锁定只读文件。

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

它功能强大，文档也相对完善。它是基于紧固件的。

其他的解决方案:

Portalocker:需要pywin32，这是一个exe安装，所以不可能通过pip 紧固件:缺乏记录 lockfile:弃用 flufl。lock:用于POSIX系统的nfs安全文件锁定。 simpleflock:上次更新2013-07 佐。lockfile:上一次更新2016-06(截至2017-03) lock_file: 2007-10年最后一次更新

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

代码如下:

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')

编辑—— 在考虑了上面关于过期锁的一些评论之后，我编辑了代码，添加了一个检查“锁文件”是否过期的检查。在我的系统上计算这个函数的几千次迭代，得到的平均值为0.002066…几秒钟前:

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

到之后:

remove('errloglock')

所以我想我将从5倍的量开始，以表示过时并监控问题的情况。

此外，当我在处理计时时，我意识到我有一些并不真正必要的代码:

lock.close()

这是我在公开声明之后立即得到的，所以我在这次编辑中删除了它。

下面是一个如何使用filelock库的例子，它类似于Evan Fossmark的实现:

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:块中的任何代码都是线程安全的，这意味着它将在另一个进程访问该文件之前完成。