本文目录导读：
Python 脚本监控
基础版（实时监控新增内容）
#!/usr/bin/env python3
import time
import os
def monitor_file(filepath, callback=None):
"""监控文件新增内容"""
# 获取文件初始大小
file_size = os.path.getsize(filepath)
print(f"开始监控 {filepath}，当前大小: {file_size} 字节")
try:
with open(filepath, 'r', encoding='utf-8') as f:
# 跳到文件末尾
f.seek(file_size)
while True:
# 读取新增内容
line = f.readline()
if line:
if callback:
callback(line)
else:
print(line, end='')
else:
# 检查文件是否被截断（变小）
current_size = os.path.getsize(filepath)
if current_size < file_size:
print(f"\n文件被截断，重置位置...")
f.seek(0)
file_size = current_size
time.sleep(0.1)
except FileNotFoundError:
print(f"文件 {filepath} 不存在")
except KeyboardInterrupt:
print("\n监控停止")
# 自定义回调函数
def process_line(line):
if 'ERROR' in line:
print(f"[错误] {time.ctime()}: {line}", end='')
if __name__ == "__main__":
monitor_file("/path/to/your/file.log", process_line)
高级版（带统计功能）
#!/usr/bin/env python3
import time
import os
import sys
from collections import Counter
from datetime import datetime
class FileMonitor:
def __init__(self, filepath):
self.filepath = filepath
self.initial_size = os.path.getsize(filepath)
self.total_lines = 0
self.keyword_counts = Counter()
self.start_time = time.time()
def monitor(self, keywords=None):
"""监控文件增长并统计"""
with open(self.filepath, 'r', encoding='utf-8') as f:
f.seek(self.initial_size)
while True:
line = f.readline()
if line:
self.total_lines += 1
self.process_line(line, keywords)
else:
self.display_stats()
time.sleep(1)
def process_line(self, line, keywords):
"""处理每一行"""
if keywords:
for keyword in keywords:
if keyword in line:
self.keyword_counts[keyword] += 1
def display_stats(self):
"""显示统计信息"""
os.system('clear') # 或 'cls' 在Windows
elapsed = time.time() - self.start_time
file_size = os.path.getsize(self.filepath)
print(f"= 文件监控报告 =")
print(f"文件: {self.filepath}")
print(f"运行时间: {elapsed:.1f} 秒")
print(f"当前文件大小: {file_size} 字节")
print(f"新增内容: {file_size - self.initial_size} 字节")
print(f"新增行数: {self.total_lines}")
print(f"当前时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
if self.keyword_counts:
print("\n关键词统计:")
for keyword, count in self.keyword_counts.most_common():
print(f" {keyword}: {count}次")
# 使用示例
if __name__ == "__main__":
monitor = FileMonitor("/var/log/syslog")
monitor.monitor(keywords=["ERROR", "WARNING", "INFO"])
Bash Shell 脚本
简单版（tail 命令）
#!/bin/bash # 监控文件增长（类似 tail -f） tail -f /var/log/syslog
增强版 Bash 脚本
#!/bin/bash
LOG_FILE="/path/to/your/file.log"
PATTERN="ERROR"
echo "开始监控 $LOG_FILE"
echo "搜索模式: $PATTERN"
# 使用 tail -f 并过滤
tail -f "$LOG_FILE" | while read line; do
# 时间戳
timestamp=$(date "+%Y-%m-%d %H:%M:%S")
# 检查是否匹配模式
if echo "$line" | grep -qi "$PATTERN"; then
echo "[$timestamp] 匹配: $line"
else
echo "[$timestamp] $line"
fi
# 可选：执行其他命令
# echo "$line" | grep "$PATTERN" >> matched_lines.log
done
带轮转处理
#!/bin/bash
LOG_FILE="/var/log/application.log"
LAST_SIZE=0
CHECK_INTERVAL=2 # 秒
# 获取初始大小
if [ -f "$LOG_FILE" ]; then
LAST_SIZE=$(stat -c%s "$LOG_FILE")
else
echo "文件不存在: $LOG_FILE"
exit 1
fi
echo "开始监控，初始大小: $LAST_SIZE 字节"
while true; do
sleep $CHECK_INTERVAL
# 检查文件是否存在
if [ ! -f "$LOG_FILE" ]; then
echo "文件被删除或轮转!"
# 等待文件重建
while [ ! -f "$LOG_FILE" ]; do
sleep 1
done
LAST_SIZE=0
echo "文件重新创建，继续监控..."
continue
fi
CURRENT_SIZE=$(stat -c%s "$LOG_FILE")
# 检查文件是否被截断（大小变小）
if [ $CURRENT_SIZE -lt $LAST_SIZE ]; then
echo "检测到日志轮转，重置位置"
LAST_SIZE=$CURRENT_SIZE
continue
fi
# 如果有新内容
if [ $CURRENT_SIZE -gt $LAST_SIZE ]; then
# 读取新增部分
tail -c $((CURRENT_SIZE - LAST_SIZE)) "$LOG_FILE"
LAST_SIZE=$CURRENT_SIZE
fi
done
Node.js 脚本
const fs = require('fs');
const readline = require('readline');
class LogMonitor {
constructor(filepath) {
this.filepath = filepath;
this.currentPosition = fs.existsSync(filepath)
? fs.statSync(filepath).size
: 0;
this.keywords = ['ERROR', 'WARNING'];
}
start() {
console.log(`开始监控 ${this.filepath}`);
// 每100ms检查一次
setInterval(() => this.checkForUpdates(), 100);
}
checkForUpdates() {
if (!fs.existsSync(this.filepath)) {
console.log('文件不存在');
return;
}
const stats = fs.statSync(this.filepath);
// 检测文件截断
if (stats.size < this.currentPosition) {
console.log('文件被截断，重置位置');
this.currentPosition = 0;
}
// 读取新增内容
if (stats.size > this.currentPosition) {
const stream = fs.createReadStream(this.filepath, {
start: this.currentPosition,
end: stats.size - 1
});
const rl = readline.createInterface({ input: stream });
rl.on('line', (line) => {
this.processLine(line);
});
rl.on('close', () => {
this.currentPosition = stats.size;
});
}
}
processLine(line) {
const timestamp = new Date().toISOString();
// 检查关键词
for (const keyword of this.keywords) {
if (line.includes(keyword)) {
console.log(`[${timestamp}] [${keyword}] ${line}`);
return;
}
}
// 或直接输出
console.log(`[${timestamp}] ${line}`);
}
}
// 使用
const monitor = new LogMonitor('/var/log/app.log');
monitor.start();
Go 语言版本
package main
import (
"fmt"
"os"
"time"
)
func main() {
filepath := "/var/log/app.log"
// 获取初始大小
fileInfo, err := os.Stat(filepath)
if err != nil {
fmt.Println("文件不存在")
return
}
lastSize := fileInfo.Size()
fmt.Printf("开始监控 %s, 初始大小: %d\n", filepath, lastSize)
for {
// 每2秒检查一次
time.Sleep(2 * time.Second)
fileInfo, err = os.Stat(filepath)
if err != nil {
fmt.Println("文件被删除")
continue
}
currentSize := fileInfo.Size()
// 检测轮转
if currentSize < lastSize {
fmt.Println("检测到日志轮转")
lastSize = 0
}
// 读取新增内容
if currentSize > lastSize {
file, _ := os.Open(filepath)
defer file.Close()
buffer := make([]byte, currentSize-lastSize)
file.Seek(lastSize, 0)
file.Read(buffer)
fmt.Printf("[%s] %s", time.Now().Format("2006-01-02 15:04:05"), buffer)
lastSize = currentSize
}
}
}
实用技巧
实时显示在Web界面
# 简单的Web监控
from flask import Flask, Response
import time
app = Flask(__name__)
@app.route('/monitor')
def stream():
def generate():
with open("/var/log/app.log", "r") as f:
f.seek(0, 2) # 跳到末尾
while True:
line = f.readline()
if line:
yield f"data: {line}\n\n"
else:
time.sleep(1)
return Response(generate(), mimetype="text/event-stream")
if __name__ == "__main__":
app.run(host="0.0.0.0", port=8080)
使用建议
- 生产环境：使用 Python 或 Go 脚本，稳定性好
- 快速调试：使用 Bash + tail -f
- 跨平台需求：Python 是最佳选择
- 高并发场景：Go 语言性能最好
- 实时统计：添加关键词计数和统计分析
选择哪种方案取决于你的具体需求（监控频率、是否需要统计、部署环境等）。
