4. 日志采集实战：文件日志读取、实时日志监控与轮转处理
日志采集是整个监控体系的起点。说白了，数据进不来，后面再漂亮的图表也是白搭。这一章我带你亲手搞定日志采集的三个核心场景：读取历史文件、模拟实时监控、处理日志轮转。嗯，都是我在生产环境里踩过坑的地方。
4.1 文件日志读取：从静态文件里捞数据
先来个最基础的——读取一个已经存在的日志文件。我个人习惯用 Python 的 open() 配合 readlines()，但这里有个坑：如果文件很大，一次性读入内存会炸。
read() 或 readlines() 全量加载。
正确的做法是逐行读取。我一般这么写：
def read_log_file(file_path):
"""逐行读取日志文件，返回生成器"""
try:
with open(file_path, 'r', encoding='utf-8') as f:
for line in f:
yield line.strip()
except FileNotFoundError:
print(f"文件 {file_path} 不存在，请检查路径")
except PermissionError:
print(f"没有权限读取 {file_path}")你看，用 yield 做成生成器，每行只占一行内存。我在项目中遇到过某次线上事故，同事用 readlines() 读了一个 2GB 的 access.log，结果服务器直接 OOM 了……从那以后，我团队里所有日志读取代码都必须用生成器。
4.2 实时日志监控：用 tail -f 的思路模拟
实时监控才是重头戏。Linux 下 tail -f 大家都会用，但 Python 里怎么实现？核心思路是：记住文件当前读取的位置，然后不断检查文件是否有新内容追加。
我封装了一个简单的实时读取器：
import time
import os
class LogTailer:
"""模拟 tail -f 的实时日志读取器"""
def __init__(self, file_path):
self.file_path = file_path
self.file = None
self._open_file()
def _open_file(self):
"""打开文件并定位到末尾"""
self.file = open(self.file_path, 'r', encoding='utf-8')
# 跳到文件末尾，模拟 tail -f 的起始行为
self.file.seek(0, os.SEEK_END)
def follow(self, interval=0.5):
"""持续监听新追加的日志行"""
while True:
line = self.file.readline()
if line:
yield line.strip()
else:
time.sleep(interval)
# 检查文件是否被轮转（后面会讲）
if self._check_rotation():
self._reopen_file()
def _check_rotation(self):
"""检查文件是否被轮转（inode 变化）"""
try:
current_inode = os.stat(self.file_path).st_ino
return current_inode != os.fstat(self.file.fileno()).st_ino
except FileNotFoundError:
return True
def _reopen_file(self):
"""重新打开文件（处理轮转）"""
self.file.close()
self._open_file()seek(0, os.SEEK_END) 这行代码很关键。它让程序从文件尾部开始读，而不是从头。这样你启动监控时，不会把历史日志再处理一遍。
为什么会用 time.sleep(interval)？你想想看，如果没有这个休眠，while True 循环会疯狂占用 CPU。我一般设 0.5 秒，既保证实时性，又不浪费资源。
4.3 日志轮转处理：躲不开的坑
日志轮转是生产环境里最容易被忽视的问题。Linux 下 logrotate 工具会定期把当前日志重命名（比如 app.log 变成 app.log.1），然后新建一个空白的 app.log。如果你的监控程序还盯着旧文件的文件描述符，那新日志就全丢了。
我曾经在凌晨 3 点被运维电话叫醒，说监控面板上数据断了两个小时。排查下来，就是日志轮转后，采集程序还在读一个已经不更新的旧文件。
解决方案就是上面代码里的 _check_rotation 方法。核心原理：
| 检测方式 | 原理 | 优点 | 缺点 |
|---|---|---|---|
| inode 对比 | 比较文件打开时的 inode 和当前文件的 inode | 准确、高效 | 需要系统调用 |
| 文件大小回退 | 检测到文件大小突然变小 | 简单、跨平台 | 可能误判（比如日志被清空） |
| 修改时间戳 | 检测文件的 mtime 变化 | 实现简单 | 不够精确 |
我个人推荐用 inode 方式。Linux 下每个文件都有唯一的 inode 号，轮转后新文件的 inode 会变。检测到变化后，关闭旧文件描述符，重新打开新文件即可。
4.4 完整实战：整合成一个采集器
把上面三个功能整合起来，就是一个能用的日志采集器了。我写了个简化版：
import time
import os
import json
class LogCollector:
"""日志采集器：支持历史读取 + 实时监控 + 轮转处理"""
def __init__(self, file_path, callback=None):
self.file_path = file_path
self.callback = callback or self.default_callback
self.tailer = LogTailer(file_path)
self.position_file = f"{file_path}.position"
def default_callback(self, line):
"""默认回调：打印到控制台"""
print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] {line}")
def collect_historical(self):
"""采集历史日志（从文件开头到当前位置）"""
print("开始采集历史日志...")
with open(self.file_path, 'r', encoding='utf-8') as f:
for line in f:
self.callback(line.strip())
print("历史日志采集完成")
def collect_realtime(self):
"""实时采集新日志"""
print("开始实时监控日志...")
for line in self.tailer.follow():
self.callback(line)
def run(self):
"""一键运行：先历史，再实时"""
self.collect_historical()
self.collect_realtime()
# 使用示例
if __name__ == "__main__":
collector = LogCollector("/var/log/myapp/access.log")
collector.run()- 先处理历史日志，再进入实时监控——保证数据不丢不重
- 回调函数机制——方便对接后续的解析、存储、告警模块
- 轮转检测内置在 tailer 中——对上层透明
4.5 知识体系总览
下面这张图把本章的核心逻辑串起来了。你看一眼就能明白整个采集流程：
嗯，这张图把整个采集流程串起来了。你从左边开始看：先读历史日志，然后进入实时监控，中间自动处理轮转，最后输出结构化数据给下游。我在实际项目中就是按这个架构搭的，稳定跑了两年没出过问题。
好了，这一章的内容就这些。日志采集是基础中的基础，但也是坑最多的地方。你把这三个场景吃透了，后面的解析和可视化才能站得住脚。
公众号:蓝海资料掘金营，微信deep3321