如何从流附加到文件而不是覆盖 Python

Question

我是 python 的新手，所以这可能是一个简单的问题，但我有一个 kafka 消费者，我从中读取消息。每次收到新消息时，它都会将之前的消息重写到 order.json 文件中，但我想改为附加它。此外，我想使用可能的某种暂停来确保消息的读取速度不超过每 1 秒一次。任何有关如何执行此操作的提示将不胜感激。这是我当前的代码

for message in consumer:
     with open('order.json', 'w') as file:
          file.write(message.value.decode('UTF-8'))

Answer 1

在循环外打开：

with open('order.json', 'w') as file:
    for message in consumer:         
          file.write(message.value.decode('UTF-8'))

或者在外面打开并使用 a 如果你要为每个追加运行:

with open('order.json', 'a') as file:
    for message in consumer:         
          file.write(message.value.decode('UTF-8'))

Answer 2

您想在 append mode 中打开您的文件。此外，您可能不希望在每条消息上都打开文件，因为这可能是一项昂贵的操作（例如，您需要在每次关闭文件时更改文件元数据，如修改时间）：

# open file in append mode, once
with open('order.json', 'a') as file:
    for message in consumer:
        file.write(message.value.decode('UTF-8'))

关于速率限制，您可以从简单的事情开始，例如：

import time

def ratelimit(it, sleep=1.0):
    for v in it:
        yield v
        time.sleep(sleep)

if __name__ == '__main__':
    for i in ratelimit(range(10)):
        print(i)

这将确保迭代器的连续值之间至少有一秒的延迟。这是一个 asciicast 显示正在运行的速率限制器。

Answer 3

以 append 模式打开文件，因为 w'（写入模式）每次存在时截断文件

for message in consumer:
    with open('order.json', 'a') as file:
        file.write(message.value.decode('UTF-8'))

如何从流附加到文件而不是覆盖 Python

How to append to a file from stream rather than overwrite in Python

python

message

file

stream