为什么你该扔掉沉重的 ELK，试试自己搭一个日志看板？
先坦白一个场景：去年我接手一个内部微服务项目，团队为了看个实时日志，硬是上了全套 ELK 栈。Elasticsearch、Logstash、Kibana 三个组件装完，光是调 JVM 堆内存就折腾了两天。后来发现，我们那个小团队每天真正盯的，不过是 ERROR 级别的报错和某个接口的响应延迟——杀鸡用了牛刀。
不是说 ELK 不好，而是对于中小型项目、个人开发者或者团队内部工具，它的部署成本和维护精力实在太高了。你需要的，往往只是一个能实时看到日志流、能高亮错误、能在出问题时喊你一声的轻量看板。
刚好，FastAPI + WebSocket 的组合特别适合干这事。FastAPI 天生支持异步，WebSocket 可以一条长连接把日志从后端推送到浏览器，延迟几乎为零。再加上 Python 自带的 logging 模块做采集，Chart.js 画几张实时图表，一个完整的日志监控看板，两百行代码内就能跑起来。
这篇文章的目标很具体：带你从零实现一个包含数据流采集、实时可视化、异常告警的日志看板。不依赖任何重量级中间件，你只需要一台能跑 Python 的机器。
技术选型：为什么是 FastAPI + WebSocket + Chart.js？
先看整体架构，其实就三层：
- 数据源：你的 Python 应用（或者多个应用）产生的日志，通过自定义的 LogHandler 实时推送到 FastAPI 服务。
- 中转层：FastAPI 服务接收日志流，用 WebSocket 广播给所有连接的客户端，同时做规则判断（比如 ERROR 级别超过 5 次/分钟就触发告警）。
- 展示层：浏览器里一个简单的 HTML 页面，通过 WebSocket 接收日志，用列表展示每一条记录，用 Chart.js 画实时频率图和级别分布饼图。
这里 FastAPI 的优势很明显：它的 WebSocket 支持是原生的，不需要像 Flask 那样额外装扩展；异步特性让它可以同时处理大量长连接而不阻塞。再加上 Pydantic 做数据校验，日志结构化的解析几乎零成本。
前端选 Chart.js 而非 ECharts，纯粹因为它更轻量、API 更直观。如果你偏好 ECharts，替换起来也就换个图表配置的事。
第一步：搭好 FastAPI 的 WebSocket 端点
先初始化项目。创建一个新目录，装依赖：
pip install fastapi uvicorn websockets
然后写一个最简单的 main.py。核心是一个 WebSocket 端点，负责接收日志并广播给所有在线客户端。这里有个设计细节：我们用了一个全局的 connected_clients 集合来管理所有 WebSocket 连接，每当有日志消息进来，就遍历这个集合逐个推送。
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from typing import List
import asyncio
app = FastAPI()
connected_clients: List[WebSocket] = []
@app.websocket("/ws/logs")
async def websocket_endpoint(websocket: WebSocket):
await websocket.accept()
connected_clients.append(websocket)
try:
while True:
# 这里等待客户端发来的消息，但我们后面会让服务端主动推送日志
data = await websocket.receive_text()
# 实际项目中，客户端可能发送心跳或控制指令
except WebSocketDisconnect:
connected_clients.remove(websocket)
很多新手会踩一个坑：WebSocket 的 receive_text() 是阻塞的，如果你在循环里只等接收，那服务端就没办法主动推送了。解决方案是——让日志推送走另一个异步任务，或者直接用 asyncio.Queue 解耦。后面我们会用到第二种方式。
第二步：把 Python logging 的输出怼进 WebSocket
这一步是整个系统的关键：如何让应用产生的每一条日志，都自动飞到浏览器？
Python 的 logging 模块允许你自定义 Handler。我们写一个 WebSocketHandler，继承 logging.Handler，在 emit 方法里把日志格式化成 JSON，然后塞进一个全局队列。FastAPI 后台再开一个协程，从队列里取数据广播给所有 WebSocket 客户端。
import logging
import json
from datetime import datetime
class WebSocketHandler(logging.Handler):
def __init__(self, log_queue: asyncio.Queue):
super().__init__()
self.log_queue = log_queue
def emit(self, record):
log_entry = {
"timestamp": datetime.fromtimestamp(record.created).isoformat(),
"level": record.levelname,
"logger": record.name,
"message": record.getMessage()
}
asyncio.ensure_future(self.log_queue.put(json.dumps(log_entry)))
然后在 FastAPI 启动时，配置 logger 并启动广播任务：
@app.on_event("startup")
async def startup():
log_queue = asyncio.Queue()
# 配置根日志器
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger()
logger.addHandler(WebSocketHandler(log_queue))
# 启动后台广播任务
asyncio.create_task(broadcast_logs(log_queue))
广播函数也很简单：
async def broadcast_logs(log_queue: asyncio.Queue):
while True:
log_json = await log_queue.get()
for client in connected_clients:
try:
await client.send_text(log_json)
except:
pass # 客户端断开时自动清理
测试一下：在任意代码里写 logging.error("数据库连接超时")，这条日志就会在几十毫秒内出现在所有打开看板的浏览器上。你可以试试同时开三个浏览器窗口，每个都能收到——这就是广播的效果。
第三步：前端实时看板，别太丑就行
前端部分我直接写在一个 HTML 文件里，简单粗暴。核心逻辑就是：连接 WebSocket，收到日志后追加到页面列表里，同时更新图表。
先写一个日志滚动列表：
const ws = new WebSocket("ws://localhost:8000/ws/logs");
const logContainer = document.getElementById("log-list");
ws.onmessage = function(event) {
const log = JSON.parse(event.data);
const div = document.createElement("div");
div.className = `log-entry log-${log.level.toLowerCase()}`;
div.textContent = `${log.timestamp} [${log.level}] ${log.message}`;
logContainer.appendChild(div);
// 自动滚动到底部
logContainer.scrollTop = logContainer.scrollHeight;
};
图表我用 Chart.js 画两个：一个是实时频率图（每 5 秒统计日志条数），一个是日志级别分布饼图。频率图的数据来自一个滑动窗口——保留最近 30 秒的日志时间戳，每 5 秒更新一次。实现起来就是维护一个数组，新日志进来 push，过期的 shift。
这里有个实用小技巧：日志级别用颜色区分——INFO 灰色，WARNING 黄色，ERROR 红色，CRITICAL 深红加粗。用户扫一眼就知道有没有问题。
第四步：告警——不能让错误日志躺在那没人管
光有看板还不够，异常必须及时通知。我们实现一个简单的规则引擎：
- 规则 1：1 分钟内 ERROR 级别日志超过 3 条，触发告警。
- 规则 2：日志消息包含关键字“OOM”或“Timeout”，立即触发。
在后端广播函数里，每收到一条日志就检查规则。如果命中，就调用告警模块。告警方式可以很多样：最简单的发邮件（用 smtplib），或者发钉钉/企业微信 Webhook。我习惯用钉钉 Webhook，因为它配置简单，而且手机端通知很及时。
async def check_alerts(log_entry: dict):
if log_entry["level"] == "ERROR":
# 滑动窗口计数
error_timestamps.append(time.time())
recent_errors = [t for t in error_timestamps if time.time() - t < 60]
if len(recent_errors) >= 3:
await send_dingtalk_alert(f"最近1分钟出现{len(recent_errors)}条ERROR日志！")
if "OOM" in log_entry["message"]:
await send_dingtalk_alert("检测到OOM关键字，请立即检查！")
前端收到告警时，我会让页面标题闪烁，并播放一段短暂的提示音（用 Web Audio API 生成一个简单的蜂鸣声，不需要音频文件）。这样即使你看别的页面，也能被拉回来。
第五步：部署时踩过的几个坑
本地跑通之后，部署到服务器时有些细节要注意。
首先，别直接用 uvicorn main:app 跑生产。用 gunicorn -k uvicorn.workers.UvicornWorker main:app，可以多进程处理请求。但注意 WebSocket 连接是粘在单个 worker 上的，如果开了多个 worker，需要确保 WebSocket 连接都路由到同一个 worker——要么用 --worker-connections 限制，要么用 Redis 做跨进程广播。
其次，WebSocket 需要心跳检测。我每 30 秒从服务端发一个 {"type": "ping"}，客户端收到后回复 {"type": "pong"}。如果连续 3 次没收到 pong，就关闭连接并清理资源。这能防止僵尸连接占用内存。
最后，日志要不要持久化？如果你需要历史查询，最简单的方案是集成 SQLite。在广播的同时，把日志写入 SQLite 表。前端加一个搜索框，可以按时间、级别、关键字查询历史日志。SQLite 对单机应用完全够用，而且零配置。
总结：从两百行代码到可用的监控工具
回头看一下，我们做的东西其实不复杂：FastAPI 提供 WebSocket 端点，Python logging 的输出通过自定义 Handler 推送到队列，后台协程广播给所有浏览器，前端用 Chart.js 画实时图表，加上简单的规则告警。全部代码加起来，核心部分不到 300 行。
如果你愿意，还能继续扩展：比如对接多个应用的日志（每个应用用不同的 logger name 区分），或者接入 OpenTelemetry 做分布式追踪。但至少现在，你已经有了一个趁手的轻量级工具，不用再为看个日志就去折腾 ELK 了。
最后丢两个链接：FastAPI 官方文档的 WebSocket 章节（写得非常清楚），以及 Chart.js 的实时图表示例。照着敲一遍，半小时内你就能跑起来。
评论