本文详解 python 中使用 `readline()` 实时监控追加写入文件时“读不到新行”的根本原因，重点说明缓冲机制的影响，并提供线程安全、低延迟的可靠实现方案。

在多线程或后台任务中实现类似 Unix tail

? 根本原因：写入端未刷新缓冲区

当另一个进程（或线程）以文本模式（如 'a' 或 'w'）打开文件并写入内容时，Python 默认启用块缓冲（block buffering）。这意味着：

• 写入的数据先暂存于内存缓冲区；
• 仅当缓冲区满、文件关闭、或显式调用 .flush() 时，数据才真正落盘；
• 因此，即使 os.path.getsize() 显示文件已增大（该函数读取的是磁盘元数据），readline() 仍可能因底层文件指针未感知新增内容而返回空。

✅ 关键结论：readline() 是否能读到新行，取决于写入方是否已将数据刷入磁盘，而非“是否调用了 write()”。

✅ 正确实践：双端协同控制缓冲行为

方案一：写入端启用行缓冲（推荐）

# 写入方（例如日志生成器）
with open("log.txt", "a", buffering=1) as f:  # buffering=1 → line-buffered
    f.write("New log entry\n")  # 自动 flush（遇 '\n' 即刷盘）

方案二：写入端手动刷新

with open("log.txt", "a", encoding="utf-8") as f:
    f.write("New log entry\n")
    f.flush()  # 强制立即写入磁盘

方案三：读取端优化（增强健壮性）

你的原始读取逻辑可升级为以下更可靠的版本：

import os
import time
import io

def tail_file(file_path, stop_event):
    with io.open(file_path, 'r', encoding='utf-8') as file:
        # 定位到当前文件末尾
        file.seek(0, 2)
        while not stop_event.is_set():
            line = file.readline()
            if line:
                print(f"LINE: {line.rstrip()}")
            else:
                # 文件无新内容 → 短暂休眠避免忙等；也可用 select/poll 做更高效等待（需 OS 支持）
                time.sleep(0.1)  # 比 1s 更灵敏，降低延迟
                # 可选：检查文件是否被截断（如 logrotate 场景）
                if file.tell() > os.path.getsize(file_path):
                    file.seek(0, 2)  # 重置到新末尾

⚠️ 注意事项与最佳实践

• 不要混用二进制与文本模式：读取端用 'r'，写入端必须用 'a' 或 'w'（文本模式），否则换行符处理不一致。
• 编码必须统一：encoding='utf-8' 需在读/写两端显式指定，避免解码错误。
• 线程安全：file.readline() 是线程安全的，但多个线程同时写同一文件需额外加锁（推荐由单一写入者负责）。
• 跨平台兼容性：buffering=1 在 Windows 和 Unix-like 系统均有效；避免依赖 os.stat().st_size 做轮询判断（精度低、开销大）。
• 生产环境建议：对于高吞吐日志，优先考虑 watchdog 库监听文件系统事件，或使用 tailer 等成熟第三方包。

通过精准控制缓冲策略与读写协同，即可构建稳定、低延迟的实时文件监控逻辑——这不仅是技术细节的修正，更是对 Python I/O 模型本质的深入理解。