在python中将线程安全写入文件答案

【问题标题】：Write thread-safe to file in python在python中将线程安全写入文件
【发布时间】：2020-06-21 11:55:48
【问题描述】：

如何在 Python 线程安全中将数据写入文件？我想为每个请求和每个小时将一些变量保存到一个文件中，我想做一些分组并将其写入 mysql。

在 Java 中，我现在将其放入缓存的数组中，当数组已满时将其写入文件。

如何在 Python 中做到这一点？并发请求很多，所以它必须是线程安全的。

编辑：

我们最终使用了运行良好的日志记录模块。

【问题讨论】：

你的意思是“线程安全”吗？
为什么不直接将行写入 MySQL，让其并发处理，然后聚合并做必要的事情？
可以使用锁定和排队等各种方式，但更容易将其扔到服务器上，在那里聚合，或者在 cronjob 或类似的任务中，在服务器中聚合并插入聚合，或者从客户端拉回数据、聚合和插入...
logging 模块是线程安全的。您可以使用它从单独的线程写入文件。
@JonClements 写入 mysql 是我想要阻止的。这需要大量资源，因为它每秒收到大约 300 个请求。

标签： python

【解决方案1】：

我们使用了日志模块：

import logging

logpath = "/tmp/log.log"
logger = logging.getLogger('log')
logger.setLevel(logging.INFO)
ch = logging.FileHandler(logpath)
ch.setFormatter(logging.Formatter('%(message)s'))
logger.addHandler(ch)


def application(env, start_response):
   logger.info("%s %s".format("hello","world!"))
   start_response('200 OK', [('Content-Type', 'text/html')])
   return ["Hello!"]

【讨论】：

+1 用于滥用logging 以线程安全地将行写入文件，我有时会这样做。有一些必需的样板和了解日志记录的工作原理，但不重新发明轮子很好。

【解决方案2】：

看看Queue 类，它是线程安全的。

from Queue import Queue
writeQueue = Queue()

在线程中

writeQueue.put(repr(some_object))

然后将其转储到文件中，

outFile = open(path,'w')
while writeQueue.qsize():
  outFile.write(writeQueue.get())
outFile.flush()
outFile.close()

Queue 将接受任何 python 对象，因此如果您尝试执行打印到文件以外的其他操作，只需通过 Queue.put 存储来自工作线程的对象。

如果您需要在脚本的多次调用中拆分提交，您将需要一种将部分构建的提交缓存到磁盘的方法。为避免多个副本同时尝试写入文件，请使用lockfile 模块，可通过 pip 获得。我通常使用 json 对数据进行编码来实现这些目的，它支持序列化字符串、unicode、列表、数字和 dicts，并且比 pickle 更安全。

with lockfile.LockFile('/path/to/file.sql'):
  fin=open('/path/to/file')
  data=json.loads(fin.read())
  data.append(newdata)
  fin.close()
  fout=open('/path/to/file','w')
  fout.write(json.dumps(data))
  fout.close()

请注意，根据操作系统功能，锁定和解锁文件以及为每个请求重写文件所花费的时间可能比您预期的要长。如果可能，请尝试仅附加到文件，因为这样会更快。此外，您可能希望使用客户端/服务器模型，其中每个“请求”都会启动一个工作脚本，该脚本连接到服务器进程并通过网络套接字转发数据。这回避了对锁定文件的需求；根据您正在谈论的数据量，它可能能够将它们全部保存在服务器进程的内存中，或者服务器可能需要将其序列化到磁盘并以这种方式将其传递给数据库。

WSGI 服务器示例：

from Queue import Queue
q=Queue()
def flushQueue():
    with open(path,'w') as f:
       while q.qsize():
           f.write(q.get())

def application(env, start_response):
   q.put("Hello World!")
   if q.qsize() > 999:
       flushQueue()
   start_response('200 OK', [('Content-Type', 'text/html')])
   return ["Hello!"]

【讨论】：

感谢您的示例，但是队列如何仍然存在于单独的请求中？我虽然每当一个 python 脚本（或任何脚本语言）完成时，就不可能使用先前请求中的任何对象/变量等？
好的，这取决于您对请求的确切含义。当解释器退出时，它使用的所有东西都会被 GCed，所以有两个选项可以处理这个问题。首先，不要让解释器退出，如果你在谈论线程安全，大多数人会假设你有某种持续运行的守护进程，它可以在请求之间保留状态。如果您的意思是您需要并行运行多个进程（解释器的副本），并且需要避免混合写入，那么操作系统级别的锁定文件就是解决方案。我也会用它来扩展我的例子。
对于线程安全，我的意思是混合写入。我正在使用Linux顺便说一句。我认为锁定和解锁文件的时间/资源很大。我们在 Java 中所做的是在每次请求后将字符串写入内存中的数组，当数组已满时，会创建一个新数组并将旧数组写入文件。根据请求的数量，我们可以增加/减少数组的大小（默认为 1000）。
Java 会遇到与 python 在 JVM 退出时数组获取 GC 相同的问题，因此它必须在请求之间持续存在。我建议对 python 做同样的事情，并使用队列。如果每次生成请求的地方都必须启动一个新的解释器，那么使用第二个进程并通过 lo 接口将数据传递给它。请求是如何产生的？你使用什么框架？
我们不使用任何框架。我想使用 nginx+gunicorn。我只编写了单个 .py 脚本文件。你能给出一个示例代码，当队列为 1000 大时，将“hello world”放入队列并写入文件？

【解决方案3】：

我制作了一个简单的编写器，它使用 threading 和 Queue 并且可以很好地处理多个线程。优点：理论上它可以接收来自多个进程的数据而不阻塞它们，并在其他线程中异步写入。缺点：额外的写作线程会消耗资源；在 CPython 中 threading 没有提供真正的多线程。

from queue import Queue, Empty
from threading import Thread

class SafeWriter:
    def __init__(self, *args):
        self.filewriter = open(*args)
        self.queue = Queue()
        self.finished = False
        Thread(name = "SafeWriter", target=self.internal_writer).start()  
    
    def write(self, data):
        self.queue.put(data)
    
    def internal_writer(self):
        while not self.finished:
            try:
                data = self.queue.get(True, 1)
            except Empty:
                continue    
            self.filewriter.write(data)
            self.queue.task_done()
    
    def close(self):
        self.queue.join()
        self.finished = True
        self.filewriter.close()
                    
#use it like ordinary open like this:
w = SafeWriter("filename", "w")
w.write("can be used among multiple threads")
w.close() #it is really important to close or the program would not end

【讨论】：