如何读取被其他程序持续写入的文件？日志解析方案咨询

你的问题非常典型——在模拟tail -f时，持续持有文件句柄确实会和日志写入进程产生冲突，尤其是当服务器需要截断或重新创建日志文件的场景。先直接给结论：你提出的「关闭重开+记录读取位置」的方案是更优的选择，而且确实需要记录上次读取的位置来定位新内容，不过还要额外处理日志文件被截断的情况。

原方案的核心问题 #

你当前的代码一直保持文件打开状态，会带来两个关键问题：

1. 句柄占用导致写入冲突：如果服务器重启时尝试以独占模式重新打开日志文件（比如某些日志系统的配置），你的程序持有的文件句柄会导致服务器无法获取写入权限，从而无法写入日志。
2. inode关联失效：当服务器将日志归档并截断原文件后，你的程序的文件句柄仍然指向原来的文件inode（即使原文件内容被清空），新写入的内容会进入新的inode（如果文件被重新创建），导致你的程序完全读不到新日志。

你的方案的可行性与优化点 #

你描述的流程（打开→读取→存位置→关闭→解析→重开→seek→读取→关闭）是完全可行的，而且能解决原方案的问题。不过需要补充一个关键判断：检查文件当前大小是否小于上次记录的位置——这是处理服务器截断日志的核心逻辑。

关键实现细节 #

• 记录读取位置：用tell()获取每次读取后的位置，存在内存中即可（如果需要程序重启后恢复位置，可以持久化到本地文件）。
• 处理文件截断：每次打开文件后，先通过os.fstat()获取文件当前大小。如果当前大小小于上次记录的位置，说明日志被截断了，需要将读取位置重置为0，重新读取新内容。
• 自动管理文件句柄：使用with语句打开文件，确保每次读取后自动关闭句柄，避免资源泄漏。
• 异常处理：捕获FileNotFoundError，因为服务器归档后可能会暂时删除原日志文件，再重新创建，这时候需要跳过当前循环，等待下一次重试。

示例代码 #

结合你的asyncio场景，优化后的代码如下：

import os
import asyncio

async def tail_log(log_path):
    last_read_pos = 0
    while True:
        try:
            with open(log_path, "rt") as log_file:
                # 获取文件当前大小，判断是否被截断
                file_stats = os.fstat(log_file.fileno())
                if file_stats.st_size < last_read_pos:
                    # 日志被截断，重置读取位置到开头
                    last_read_pos = 0
                # 定位到上次读取的位置
                log_file.seek(last_read_pos)
                # 读取所有新内容
                new_content = log_file.readlines()
                if new_content:
                    # 处理日志内容（替换成你的业务逻辑）
                    for line in new_content:
                        print(f"New log line: {line.strip()}")
                    # 更新最后读取位置
                    last_read_pos = log_file.tell()
        except FileNotFoundError:
            # 日志文件可能被归档删除，短暂等待后重试
            pass
        # 控制轮询间隔，避免频繁IO
        await asyncio.sleep(1)

# 启动异步任务
asyncio.run(tail_log("log.txt"))

额外说明 #

• 这种方案的性能开销很小，因为每次打开/关闭文件的操作在现代系统中是轻量级的，加上1秒的轮询间隔，不会对系统造成负担。
• 如果你需要更高效的方式（比如避免轮询），可以考虑使用Linux的inotify机制（通过pyinotify库）来监听文件变化，但这会增加代码的复杂度，且跨平台性较差。你的轮询方案已经能满足大多数场景的需求。

内容的提问来源于stack exchange，提问作者Onyxdragun