基于Python的Spring Boot慢SQL日志分析工具
自动化慢SQL检测与分析系统
本工具专为Java Spring Boot应用中的MyBatis日志设计，可自动识别执行耗时超过阈值的数据库操作，并生成结构化报告。
核心功能特性
- 支持标准Spring Boot MyBatis日志格式解析
- 动态配置慢查询触发阈值（毫秒级）
- 多线程环境下的SQL执行记录关联匹配
- 生成可导入表格的CSV结果文件
- 实时进度反馈与性能统计摘要
输入日志格式要求
14:23:15.456 [task-thread-7] DEBUG UserMapper.selectById - ==> Preparing: SELECT id, name FROM users WHERE id = ?
14:23:15.458 [task-thread-7] DEBUG UserMapper.selectById - ==> Parameters: 123(Integer)
14:23:18.902 [task-thread-7] DEBUG UserMapper.selectById - <== Total: 1
快速使用指南
直接运行脚本并指定日志路径：
python analyze_slow_sql.py /var/log/app/application.log -t 3000 -o findings.csv
命令行参数说明
| 参数 | 描述 | 默认值 |
|---|---|---|
log_file | 待分析的日志文件路径 | 必需 |
-o, --output | 输出CSV文件位置 | auto-generated |
-t, --threshold | 慢查询判定阈值（毫秒） | 5000 |
输出内容详解
控制台将展示如下信息：
============================================================
慢SQL分析汇总
============================================================
总执行次数: 45,231 次
慢查询数量: 1,347 次
占比: 2.98%
最高耗时: 87,654.00ms
最慢语句: SELECT * FROM orders o JOIN customers c ON o.cust_id = c.id WHERE o.status = ? AND c.region = ? ...
导出的CSV包含以下字段：
- 线程标识：执行上下文名称
- Mapper类名：数据访问接口全限定名
- 原始SQL：完整查询语句
- 绑定参数：实际传入的变量值
- 返回行数：查询结果集大小
- 执行时长：从准备到完成的时间差
- 起止时间戳：精确到毫秒
- 阈值基准：设定的判断标准
- 日志行号：原始日志中对应行的位置
底层工作流程
- 逐行扫描：以流式方式读取大文件，避免内存溢出
- 正则匹配：通过预编译模式识别三类日志事件
- 上下文聚合：根据线程名建立同一请求的完整生命周期
- 时间计算：基于开始和结束时间点精确测算耗时
- 智能筛选：仅保留超出阈值的记录进行持久化
最佳实践建议
- 确保日志级别设置为DEBUG或TRACE
- 确认日志编码为UTF-8，避免乱码问题
- 对于超大日志（>100MB），建议分段处理
- 使用
-t参数根据业务场景调整敏感度
故障排查提示
- 若无结果，请检查日志是否包含完整的
Preparing→Total链路 - 出现编码异常时，可用文本编辑器转换为UTF-8格式
- 如需验证正则匹配逻辑，可在代码中临时启用调试打印
扩展能力预留
当前架构支持后续增强：
- 批量处理多个日志文件
- 按时间段过滤特定窗口内的操作
- 按语句类型（SELECT/INSERT/UPDATE）分类统计
- 生成时间趋势图用于性能监控
源码实现概览
以下是关键部分的重构版本：
import re
import csv
from datetime import datetime
from argparse import ArgumentParser
class QueryAnalyzer:
def __init__(self, threshold_ms=5000):
self.threshold = threshold_ms
self.records = {}
self.slow_queries = []
def parse_timestamp(self, ts_str):
try:
return datetime.strptime(ts_str, '%H:%M:%S.%f')
except:
return None
def extract_event(self, line):
# 统一正则表达式处理不同类型的日志条目
patterns = [
(r'(\d{2}:\d{2}:\d{2}\.\d{3})\s+\[([^\]]+)\]\s+(\w+)\s+([^\s]+)\.(\w+)\s+-\s*==>\s*Preparing:\s*(.+)',
'PREPARE'),
(r'(\d{2}:\d{2}:\d{2}\.\d{3})\s+\[([^\]]+)\]\s+(\w+)\s+([^\s]+)\.(\w+)\s+-\s*==>\s*Parameters:\s*(.+)',
'PARAMS'),
(r'(\d{2}:\d{2}:\d{2}\.\d{3})\s+\[([^\]]+)\]\s+(\w+)\s+([^\s]+)\.(\w+)\s+-\s*<==\s+Total:\s*(\d+)',
'RESULT')
]
for pattern, event_type in patterns:
match = re.match(pattern, line.strip())
if match:
return {
'type': event_type,
'time': match.group(1),
'thread': match.group(2),
'level': match.group(3),
'mapper': match.group(4),
'method': match.group(5),
'content': match.group(6) if len(match.groups()) > 6 else ''
}
return None
def process_log(self, log_path):
with open(log_path, 'r', encoding='utf-8') as f:
for line_num, line in enumerate(f, start=1):
parsed = self.extract_event(line)
if not parsed:
continue
thread_key = f"{parsed['thread']}_{parsed['mapper']}_{parsed['method']}"
if parsed['type'] == 'PREPARE':
self.records[thread_key] = {'prepare': parsed}
elif parsed['type'] == 'PARAMS':
if thread_key in self.records:
self.records[thread_key]['params'] = parsed
elif parsed['type'] == 'RESULT':
if thread_key in self.records and 'prepare' in self.records[thread_key]:
prep_time = self.parse_timestamp(self.records[thread_key]['prepare']['time'])
result_time = self.parse_timestamp(parsed['time'])
duration = (result_time - prep_time).total_seconds() * 1000
if duration >= self.threshold:
self.slow_queries.append({
'sql': self.records[thread_key]['prepare']['content'],
'duration': duration,
'rows': int(parsed['content']),
'thread': parsed['thread'],
'timestamp': result_time.strftime('%H:%M:%S.%f')[:-3]
})
del self.records[thread_key]
def export_results(self, output_path):
if not self.slow_queries:
print("未发现慢查询")
return
with open(output_path, 'w', encoding='utf-8-sig', newline='') as f:
writer = csv.DictWriter(f, fieldnames=['SQL', '耗时(ms)', '返回行数', '线程', '时间'])
writer.writeheader()
for q in sorted(self.slow_queries, key=lambda x: x['duration'], reverse=True):
writer.writerow(q)
def main():
parser = ArgumentParser(description='Spring Boot慢查询分析工具')
parser.add_argument('log_file', help='日志文件路径')
parser.add_argument('-t', '--threshold', type=int, default=5000)
parser.add_argument('-o', '--output', default='slow_queries.csv')
args = parser.parse_args()
analyzer = QueryAnalyzer(threshold_ms=args.threshold)
analyzer.process_log(args.log_file)
analyzer.export_results(args.output)
if __name__ == '__main__':
main()
