23 级山东大学软件学院创新实训 - 个人纪录（七）
·
LinguaSpark 请求响应与 SQL 日志系统的设计与实现
在 LinguaSpark 智能外语学习平台从开发阶段向生产环境过渡的过程中，我发现系统缺乏有效的可观测性手段——当用户反馈接口异常或数据错误时，开发人员需要登录服务器手动查看日志，排查效率极低。为此，我设计并实现了一套完整的请求响应与 SQL 日志系统，包括基于中间件的请求日志自动采集、基于 SQLAlchemy 事件的 SQL 语句捕获、基于文件的异步日志存储引擎，以及配套的 Web 可视化日志查看器和 RESTful API。整个系统零外部依赖，与 FastAPI 框架深度集成，开发人员无需登录服务器即可实时追踪每一次 API 调用的完整链路。
一、核心问题与需求分析
1.1 问题背景
LinguaSpark 项目在开发和测试阶段面临以下痛点：
| 痛点 | 具体表现 | 影响 |
|---|---|---|
| 接口异常难追踪 | 前端报错后无法快速定位后端哪个接口出问题 | 调试时间从分钟级变成小时级 |
| SQL 性能黑洞 | 页面卡顿但不知道是哪条 SQL 慢 | 无法做针对性优化 |
| 数据一致性问题 | 用户反馈数据不对，但不知道是哪次请求改错了 | 排查方向缺失 |
| 多服务调试困难 | 多个接口并发调用时，请求和 SQL 交织在一起 | 无法建立请求→SQL 的因果链路 |
1.2 需求分析
基于以上痛点，我梳理了日志系统的核心需求：
┌────────────────────────────────────────────────────────────┐
│ 日志系统需求全景 │
├──────────────────┬──────────────────┬──────────────────────┤
│ 请求日志 │ SQL 日志 │ 查看与检索 │
├──────────────────┼──────────────────┼──────────────────────┤
│ • 记录请求方法/路径 │ • 记录原始 SQL 语句│ • Web 可视化查看器 │
│ • 记录请求体/参数 │ • 记录绑定参数 │ • RESTful API 接口 │
│ • 记录响应状态码 │ • 记录执行耗时 │ • 按时间范围过滤 │
│ • 记录响应耗时 │ • 关联请求 ID │ • 按请求 ID 追踪 │
│ • 记录响应体(可截断)│ • 捕获 SELECT 等 │ • 分页加载 │
│ • 生成唯一请求 ID │ │ • 合成请求+SQL 视图 │
└──────────────────┴──────────────────┴──────────────────────┘
1.3 设计原则
- 零外部依赖：不引入 ELK / Grafana / Loki 等重量级组件，纯 Python 实现
- 异步非阻塞：日志写入不能拖慢正常请求响应
- 与框架深度集成：利用 FastAPI 中间件和 SQLAlchemy 事件机制，无侵入
- 日志自动轮转：按日期命名，自动清理过期日志，无需人工维护
二、系统架构设计
2.1 整体架构
┌─────────────────────────────────────────────────────────────────┐
│ FastAPI Application │
│ │
│ ┌─────────────┐ ┌──────────────────┐ ┌────────────────┐ │
│ │ HTTP 请求 │───▶│ RequestLogMiddleware│───▶│ Route Handler │ │
│ │ │ │ (请求日志中间件) │ │ (业务逻辑) │ │
│ └─────────────┘ └────────┬─────────┘ └───────┬────────┘ │
│ │ │ │
│ ▼ ▼ │
│ ┌─────────────────┐ ┌──────────────────┐ │
│ │ request_logger │ │ sql_logger │ │
│ │ (请求日志引擎) │ │ (SQL日志引擎) │ │
│ └────────┬────────┘ └────────┬─────────┘ │
│ │ │ │
│ ▼ ▼ │
│ ┌────────────────────────────────────────┐ │
│ │ AsyncLogWriter │ │
│ │ (队列 + 守护线程 + 批量写入) │ │
│ └────────────────┬───────────────────────┘ │
│ │ │
└─────────────────────────────────────┼────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────┐
│ 日志文件系统 │
│ │
│ logs/ │
│ ├── requests.log ← 请求日志（JSON 格式） │
│ └── sql_details.log ← SQL 日志（JSON 格式） │
│ │
│ logs/archive/ ← 归档目录（自动轮转） │
│ ├── requests_2026-05-01.log │
│ ├── sql_details_2026-05-01.log │
│ └── ... │
└─────────────────────────────────────────────────────────────────┘
2.2 三大核心模块
| 模块 | 文件 | 职责 |
|---|---|---|
| SQL 日志引擎 | app/core/sql_logger.py |
基于 SQLAlchemy 事件机制，拦截所有 SQL 语句并记录 |
| 请求日志引擎 | app/core/request_logger.py |
基于 FastAPI 中间件，拦截所有 HTTP 请求/响应 |
| 日志查看器 | app/core/log_viewer.py |
提供 Web UI 和 RESTful API，支持按时间/请求 ID 检索 |
三、核心模块实现详解
3.1 异步日志写入引擎（AsyncLogWriter）
这是整个日志系统的基础设施层。日志写入不能阻塞业务请求，因此我设计了一个基于 队列 + 守护线程 的异步写入器：
设计要点：
┌─────────────────────────────────────────────────────┐
│ │
│ 业务线程 守护线程 │
│ ┌──────┐ queue.put() ┌──────────────┐ │
│ │ 请求1 │─────────────────▶│ │ │
│ └──────┘ │ AsyncLog │ │
│ ┌──────┐ queue.put() │ Writer │──▶ 文件 │
│ │ 请求2 │─────────────────▶│ (单线程) │ │
│ └──────┘ │ │ │
│ ┌──────┐ queue.put() │ • 批量刷盘 │ │
│ │ 请求3 │─────────────────▶│ • 自动轮转 │ │
│ └──────┘ │ • 过期清理 │ │
│ └──────────────┘ │
│ 特点： │
│ • queue.put() 是 O(1) 操作，几乎零延迟 │
│ • 单线程消费，无需加锁 │
│ • 缓冲区 1000 条或 2 秒超时统一刷盘 │
│ • 应用退出时通过 atexit 钩子确保队列清空 │
└─────────────────────────────────────────────────────┘
核心实现逻辑：
class AsyncLogWriter:
"""异步日志写入器：队列 + 守护线程"""
def __init__(self, log_dir, filename, max_days=30):
self.queue = queue.Queue()
self.batch_buffer = []
self.batch_size = 1000 # 批量大小
self.flush_interval = 2.0 # 最大缓冲间隔（秒）
self._running = True
# 启动守护线程
self._thread = threading.Thread(target=self._write_loop, daemon=True)
self._thread.start()
# 注册退出钩子
atexit.register(self._shutdown) # 安全退出：清空队列
def write(self, entry: dict):
"""非阻塞写入（O(1) 操作）"""
self.queue.put(json.dumps(entry, ensure_ascii=False))
def _write_loop(self):
"""守护线程：批量消费队列"""
last_flush = time.time()
while self._running or not self.queue.empty():
try:
line = self.queue.get(timeout=0.5)
self.batch_buffer.append(line)
except queue.Empty:
pass
# 触发刷盘：缓冲区满 或 超时
if len(self.batch_buffer) >= self.batch_size or \
(self.batch_buffer and time.time() - last_flush >= self.flush_interval):
self._flush()
last_flush = time.time()
3.2 SQL 日志引擎
SQL 日志通过 SQLAlchemy 的 before_cursor_execute 事件来拦截所有 SQL 语句。这是整个系统中最精妙的部分——它完全不需要修改任何业务代码，只需在应用启动时注册事件监听器：
工作原理：
┌──────────────────────────────────────────────────────────┐
│ │
│ SQLAlchemy Engine │
│ ┌────────────┐ before_cursor_execute 事件 │
│ │ Session │──────────────────────────────┐ │
│ │ .execute()│ ▼ │
│ └────────────┘ ┌──────────────────┐ │
│ │ _sql_log_handler│ │
│ │ (事件回调) │ │
│ └────────┬─────────┘ │
│ │ │
│ ┌────────────────────────┘ │
│ ▼ │
│ 记录内容： │
│ ┌─────────────────────────────────────┐ │
│ │ { │ │
│ │ "request_id": "req_a1b2c3", │ ← 关联请求 │
│ │ "timestamp": "2026-06-17 10:30:01", │
│ │ "sql": "SELECT * FROM users WHERE id=%s", │
│ │ "params": [42], │
│ │ "duration_ms": 12.5 │ ← 执行耗时 │
│ │ } │ │
│ └─────────────────────────────────────┘ │
│ │
└──────────────────────────────────────────────────────────┘
request_id 的传递是整个链路追踪的核心。我使用 Python 的 contextvars 来实现跨协程/线程的上下文传递：
import contextvars
# 请求级别的上下文变量（协程安全）
_request_id_ctx: contextvars.ContextVar[str] = contextvars.ContextVar(
'request_id', default='unknown'
)
def set_request_id(request_id: str):
"""设置当前请求的 ID（在中间件中调用）"""
_request_id_ctx.set(request_id)
def get_request_id() -> str:
"""获取当前请求的 ID（在 SQL 事件回调中调用）"""
return _request_id_ctx.get()
为什么选择 contextvars 而不是 threading.local？
| 方案 | 协程安全性 | FastAPI 兼容性 | 说明 |
|---|---|---|---|
threading.local |
❌ | ❌ | 同一线程内多个协程共享，会串数据 |
contextvars |
✅ | ✅ | 每个协程独立上下文，天然隔离 |
3.3 请求日志中间件
请求日志中间件采用 纯 ASGI 中间件 实现（而非 BaseHTTPMiddleware），以获得更好的性能和更精确的响应体捕获能力：
请求生命周期与日志记录点：
时间轴 ──────────────────────────────────────────────────────▶
┌──────────────┐
│ 请求到达 │ → 记录请求开始时间
│ 生成 request_id│ → 设置 contextvars
│ 读取请求体 │ → 记录 method, path, body
└──────┬───────┘
│
▼
┌──────────────┐
│ 业务处理 │ → SQL 日志通过 contextvars 自动关联 request_id
│ (Route Handler)│
└──────┬───────┘
│
▼
┌──────────────┐
│ 响应返回 │ → 记录 status_code, 耗时
│ 捕获响应体 │ → 记录 response_body (截断 2000 字符)
│ 写入日志文件 │ → 异步写入
└──────────────┘
关键实现——响应体捕获的"偷梁换柱"技巧：
async def __call__(self, scope, receive, send):
if scope["type"] != "http":
await self.app(scope, receive, send)
return
request_id = str(uuid.uuid4())[:8]
set_request_id(request_id)
start_time = time.time()
# 存储响应信息
response_info = {"status_code": 0, "body_chunks": []}
async def capture_send(message):
"""拦截 send 调用来捕获响应数据"""
if message["type"] == "http.response.start":
response_info["status_code"] = message["status"]
elif message["type"] == "http.response.body":
response_info["body_chunks"].append(message.get("body", b""))
await send(message) # 继续正常发送，不影响业务
try:
await self.app(scope, receive, capture_send)
finally:
duration_ms = (time.time() - start_time) * 1000
# 异步写入日志，不阻塞
request_logger.log(request_id, method, path,
status_code, duration_ms,
request_body, response_body)
3.4 日志查看器
日志查看器提供了两种访问方式：Web UI 页面和 RESTful API。
┌─────────────────────────────────────────────────────────┐
│ 日志查看器架构 │
│ │
│ ┌──────────────────┐ ┌──────────────────┐ │
│ │ Web UI 页面 │ │ RESTful API │ │
│ │ /logs │ │ /api/logs/* │ │
│ └────────┬─────────┘ └────────┬─────────┘ │
│ │ │ │
│ └───────────┬───────────┘ │
│ ▼ │
│ ┌───────────────────────┐ │
│ │ 日志解析引擎 │ │
│ │ • 按时间范围过滤 │ │
│ │ • 按 request_id 追踪 │ │
│ │ • 合并请求+SQL 视图 │ │
│ │ • 分页加载 │ │
│ └───────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────┘
API 设计：
| 接口 | 方法 | 路径 | 功能 |
|---|---|---|---|
| Web 页面 | GET | /logs |
返回日志查看器 HTML 页面 |
| 请求日志列表 | GET | /api/logs/requests |
获取请求日志（支持时间范围、分页） |
| 请求详情 | GET | /api/logs/requests/{request_id} |
获取某次请求的详情 + 关联的所有 SQL |
| SQL 日志列表 | GET | /api/logs/sql |
获取 SQL 日志（支持时间范围、分页） |
Web UI 页面特点：
- 纯原生 HTML + CSS + JavaScript，无框架依赖
- 内置
fetchAPI 调用后端接口 - 支持按时间范围筛选
- 表格展示，带状态码颜色标注（2xx 绿色，4xx 黄色，5xx 红色）
- 点击请求可展开查看关联的 SQL 语句
四、关键设计决策与权衡
4.1 为什么不用标准库 logging 模块？
| 维度 | logging 模块 | 自建 JSON 日志 |
|---|---|---|
| 结构化查询 | 需额外解析文本 | JSON 原生支持字段查询 |
| 与 Web 查看器集成 | 需要复杂解析器 | 直接 json.loads |
| 多行 SQL 记录 | 需要转义处理 | JSON 字符串天然支持 |
| 性能 | 格式化开销大 | 直接序列化，更快 |
| 配置复杂度 | Handler/Formatter/Filter 链 | 单一写入器，简单 |
4.2 为什么用 contextvars 而不是全局变量？
FastAPI 使用 asyncio 异步模型，同一个线程内可能同时处理多个请求。如果使用全局变量或 threading.local 存储 request_id，在协程切换时会导致 request_id 串位——请求 A 的 SQL 被标记为请求 B 的 ID。
contextvars 是 Python 3.7+ 引入的标准库模块，专为解决此问题设计——每个 asyncio.Task 拥有独立的上下文副本，完美隔离。
4.3 为什么选择守护线程而非 asyncio 协程？
# 为什么不用 asyncio.Queue？
# 问题：SQLAlchemy 事件回调是同步的，无法直接 await
# 如果用 asyncio.Queue：
async def async_write(entry):
await async_queue.put(entry) # 需要 await
def sql_event_handler(...): # SQLAlchemy 回调是同步的！
# 没办法在这里 await
asyncio.create_task(async_write(entry)) # 创建新任务，有开销
选择守护线程的原因：
- SQLAlchemy 事件回调是同步函数，无法直接使用
await - 守护线程自动跟随主进程退出，无需手动管理生命周期
- 文件写入本身是 I/O 操作，线程模型足够高效
4.4 批量写入 vs 逐条刷盘
| 方式 | 吞吐量 | 延迟 | 数据安全 |
|---|---|---|---|
| 逐条写入 | 低（~100 QPS） | 低 | 高 |
| 批量 1000 条 | 高（~10000 QPS） | 最多 2 秒 | 中（极端情况下丢失 2 秒数据） |
| 批量 1000 条 + atexit | 高 | 最多 2 秒 | 高（正常退出不丢失） |
综合考虑，批量写入 + atexit 钩子是最优解——性能足够，正常退出不丢数据。
五、改进前后对比
| 对比维度 | 改进前 | 改进后 |
|---|---|---|
| 请求追踪 | 无 | 每个请求有唯一 ID，可追踪完整链路 |
| SQL 可见性 | 仅数据库慢查询日志 | 每条 SQL 记录原始语句+参数+耗时 |
| 请求-SQL 关联 | 无法关联 | 通过 request_id 精确关联 |
| 问题定位时间 | 30 分钟 ~ 2 小时（需登录服务器） | 30 秒（浏览器打开 Web 查看器） |
| 日志查询 | 命令行 grep / tail |
Web UI 按时间/ID 搜索 |
| 外部依赖 | — | 零（纯 Python 实现） |
| 对业务性能影响 | — | 几乎为零（异步队列，批量写入） |
| 日志存储 | 无系统化管理 | 按天轮转 + 自动清理 30 天前日志 |
六、工程实现要点
6.1 技术栈
- 后端框架：FastAPI（Python 3.10+）
- ORM：SQLAlchemy（事件机制）
- 异步写入：
queue.Queue+threading.Thread（守护线程） - 上下文传递：
contextvars（协程安全） - 日志格式：JSON Lines（每行一条 JSON）
- 前端：原生 HTML + CSS + JavaScript（零框架）
6.2 文件结构
app/core/
├── sql_logger.py # SQL 日志引擎（~130 行）
├── request_logger.py # 请求日志引擎（~150 行）
├── log_viewer.py # 日志查看器 API + Web UI（~280 行）
└── middleware.py # 认证中间件（集成 request_id 设置）
logs/
├── requests.log # 请求日志
└── sql_details.log # SQL 日志
6.3 集成方式
在 app/main.py 中仅需 3 行代码即可完成集成：
from app.core.sql_logger import setup_sql_logging
from app.core.request_logger import RequestLogMiddleware
from app.core.log_viewer import router as log_viewer_router
# 1. 注册 SQL 日志引擎（挂钩 SQLAlchemy 事件）
setup_sql_logging(engine)
# 2. 注册请求日志中间件（ASGI 中间件，最外层）
app.add_middleware(RequestLogMiddleware)
# 3. 注册日志查看器路由
app.include_router(log_viewer_router)
6.4 日志示例
请求日志（logs/requests.log）：
{"request_id":"a1b2c3d4","timestamp":"2026-06-17 10:30:01.234","method":"POST","path":"/api/words/user-words/update","query_string":"","client_ip":"127.0.0.1","request_body":"{\"word_id\": 1, \"status\": 1}","status_code":200,"duration_ms":45.2,"response_body":"{\"message\":\"复习记录更新成功\",\"interval_days\":6,...}"}
SQL 日志（logs/sql_details.log）：
{"request_id":"a1b2c3d4","timestamp":"2026-06-17 10:30:01.256","sql":"SELECT word_user.id, word_user.word_id, word_user.user_id FROM word_user WHERE word_user.word_id = %s AND word_user.user_id = %s","params":[1,42],"duration_ms":3.1}
{"request_id":"a1b2c3d4","timestamp":"2026-06-17 10:30:01.278","sql":"UPDATE word_user SET interval_days=%s, ease_factor=%s, repetitions=%s, next_review_date=%s, last_review_date=%s, last_status=%s WHERE word_user.id = %s","params":[6,250,2,"2026-06-23","2026-06-17",1,99],"duration_ms":4.7}
可以看到，同一个 request_id 将一次请求的所有 SQL 串联起来，实现了端到端的链路追踪。
以下是vibe coding主要的提示词
需求清单
- 完整记录每次接口请求、对应返回响应
- 统计接口整体响应耗时（适配慢接口场景）
- 额外采集本次请求内执行的 SQL 语句，记录单条 SQL 执行耗时
- 方案包含：数据存储、埋点采集、落地代码思路、日志查询使用说明
openEuler 是由开放原子开源基金会孵化的全场景开源操作系统项目，面向数字基础设施四大核心场景（服务器、云计算、边缘计算、嵌入式），全面支持 ARM、x86、RISC-V、loongArch、PowerPC、SW-64 等多样性计算架构
更多推荐
9
0- 0
已为社区贡献1条内容
所有评论(0)