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


当前回答

这招对我很管用: 不占用大文件,分布在几个小文件中 创建文件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)

其他回答

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

尽管正如Kevin所说,从多个进程同时写入文件是您希望尽可能避免的事情。

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

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

如果你只需要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)

锁定是平台和设备特定的,但一般来说,你有几个选择:

Use flock(), or equivalent (if your os supports it). This is advisory locking, unless you check for the lock, it's ignored. Use a lock-copy-move-unlock methodology, where you copy the file, write the new data, then move it (move, not copy - move is an atomic operation in Linux -- check your OS), and you check for the existence of the lock file. Use a directory as a "lock". This is necessary if you're writing to NFS, since NFS doesn't support flock(). There's also the possibility of using shared memory between the processes, but I've never tried that; it's very OS-specific.

对于所有这些方法,您必须使用自旋锁(失败后重试)技术来获取和测试锁。这确实为错误同步留下了一个小窗口,但它通常小到不会成为一个大问题。

如果您正在寻找跨平台的解决方案,那么您最好通过其他机制将日志记录到另一个系统(其次是上面的NFS技术)。

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

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

代码如下:

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

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