尧图网页设计 
写爬虫的人手机里大概都存着两样东西一份爬虫日志还有一个半夜响起来的报警电话。我见过太多同事凌晨两三点爬起来打开终端盯着日志滚动只为确认采集任务到底卡在哪个环节。数据没抓到还算小事错过窗口期、被目标网站识别成异常流量、又或者因为一次静默小异常导致整个任务失败那才真正让人头疼。今天想聊的这套系统就是专门解决这个问题的——一个可复现的Python爬虫运行报告监控系统。它的核心思路是给爬虫装上“行车记录仪”和“仪表盘”让每一次运行都有迹可循、有数可查、有问题能报警而且整个方案足够轻量不依赖重型运维组件。这套系统更像是一个贴合爬虫场景的轻量级埋点框架加报告中心而不是传统意义上的监控平台。你不用部署 Zabbix、Prometheus 全家桶只需要在爬虫代码的关键节点埋入监控点把请求状态、耗时、结果、异常记录到本地存储再由报告模块自动生成运行报告和告警通知。单机脚本可以接分布式采集集群也可以平移到统一观测。适合的读者很明确写过一段时间爬虫、开始觉得“跑是能跑但跑得怎么样完全靠猜”的人以及需要定期向团队或客户展示采集任务稳定性的人。1. 方案设计可复现和监控到底在解决什么问题先聊聊“可复现”这个词。很多爬虫项目本质上是个黑盒跑完了就是跑完了失败了你也只知道失败了。但监控系统要解决的是“为什么这次失败”“上次成功时候的状态是什么”“同一个请求在前后两次运行中表现有什么差异”。没有这些信息谈优化就是盲人摸象。1.1 可复现的三个层面我把可复现拆成三个层面来理解。第一层是运行过程可复现。每次爬虫任务启动时生成一个全局唯一的 run_id任务里的每个 HTTP 请求再生成一个 request_id。以后无论什么时候翻数据库只要拿到这个 ID就能把整次任务的完整轨迹拉出来——包括请求了哪些 URL、每个请求耗时多少、返回什么状态、是否被反爬拦截、最终入了几条数据。这相当于飞行记录黑匣子。第二层是问题场景可复现。当异常发生时不再是日志里孤零零的一行报错而是带着上下文当时请求的 URL 是什么、携带了哪些参数、从发起请求到抛出异常经历了多少秒、错误堆栈有哪些关键帧。有了这些上下文你才有可能用一个最小脚本把线上异常复原出来然后在本地修修补补而不是靠“听同事描述”来猜问题。第三层是结果可复现。同一套代码、同一个种子 URL、同一份配置前后跑两次结果应该能对齐。如果哪次结果明显异常你能通过报告里的指标快照一眼看出差异点在哪个环节。这有点像做菜时写好菜谱配方不变、火候不变出品就该稳定万一哪天锅糊了你至少能查出是哪一步火大了。1.2 为什么我不推荐直接用 Zabbix 或 Prometheus自己做过运维或者接过公司监控体系的读者可能会有疑问现在开源监控工具这么多直接接一套不是更省事我的观点是传统监控系统解决的是“机器健康”问题而爬虫需要的是“业务过程”观测。Zabbix、Prometheus 这类工具擅长采集 CPU、内存、带宽、进程数这些主机指标也能做告警但它们对“这次爬虫任务里有多少请求被反爬拦截了”“接口响应平均耗时多少、95分位耗时多少”“解析阶段和请求阶段各花了几秒”这类业务级信息整理起来要绕很大一圈。你可能要自定义 exporter、修改告警规则、做一堆额外开发最后得出的报表还不够直白。反过来如果你自己维护一层轻量埋点字段完全由你说了算报告长什么样也由你设计。等爬虫平台真的发展到多机分布式、日请求量百万以上再把同一套事件模型平移到 Kafka ClickHouse也完全顺理成章。前期没必要一上来就背一个重型框架。1.3 整体模块划分从埋点到报告的一整条链路这套监控系统的链路不算复杂拆开看就五个环节采集、存储、报告、告警、可视化。采集层负责在爬虫代码里埋点通过一个上下文管理器把 run 和 request 的执行过程记录成结构化事件。存储层默认使用 SQLite不引入外部服务单机环境下足够稳定。报告层定时或按需从库里聚合数据输出 JSON 摘要、HTML 报告和图表。告警层在发现成功率下跌、耗时超标、任务连续失败时通过企业微信、钉钉或邮件发通知并且在墙上留一个“手动确认”的按钮。可视化层则通过定时刷新报告页面或接入 BI 工具让团队随时能打开看。各层的具体职责可以看这张表模块核心职责落地方式采集层埋点记录 run / request / item 事件Python contextmanager侵入性小存储层保存事件明细与运行状态SQLite WAL 模式后续可换 MySQL报告层聚合指标、生成摘要与图表Python 脚本 matplotlib输出 HTML/JSON告警层阈值判断、发送通知、支持确认自定义规则 Webhook / SMTP可视化层让团队随时查看运行概况定时刷新 HTML 报告或 BI 接入这条链路建好后日常维护爬虫的心态会明显不一样不再是“默默祈祷它不要挂”而是“挂了也知道挂在哪、为什么挂、下次怎么不挂”。2. 核心细节解析埋点模型、运行指纹与存储选址在看代码之前有几个设计上的关键点需要先讲明白。这些点决定了整套系统好不好用、能不能扩展到分布式场景也决定了你在接进现有爬虫时会不会想骂人。2.1 把爬虫运行拆成四级事件模型事件模型的划分直接决定了报告的粒度。我这边的习惯是拆成 run、task、request、item 四级。run 表示一次完整的任务执行比如“2025-06-01 的商品详情页全量采集”它有开始时间、结束时间、最终状态、总请求数、成功数、失败数和入库条数。task 表示一类采集任务比如“商品详情页采集任务”“评论采集任务”方便你按业务归类多个 run。request 表示一次真实的 HTTP 请求记录 URL、请求方法、耗时、状态、错误信息以及反爬信号。item 表示一条解析后成功入库的数据记录数据的主键标识方便对比“请求了 1000 次但最终只有 800 条有效数据”这类偏差。这样拆分的原因很直接日常排查顺序是“哪类任务挂了 - 哪次运行挂了 - 哪批请求丢了 - 哪条数据错了”。四级事件正好支撑这样的逐层钻取将来做分布式也天然适配——不同 worker 采集的事件都带同一个 run_id合并时按 run_id 分区即可。2.2 运行指纹让每次请求都“查得到、可重放”这里的指纹有两类。一个是 run_id负责标识一次任务执行直接用 UUID 就够了不用纠结。另一个是 request_id用来标识一个业务意义上的“重复请求”它的生成要仔细设计。我把 request_id 设计成对请求内容做哈希的结果请求方法 标准化后的 URL 排序后的查询参数 一个时间窗口。时间窗口是关键比如 5 分钟内发起的同一个请求算同一个 ID超过 5 分钟就重新生成。这样设计有两个好处一是天然支持去重重试同一条 URL 不会产生一堆重复记录二是允许一定时间后重新请求因为反爬策略触发后往往需要等一会儿再补采。这里有个细节很坑如果你把精确到秒的当前时间也放进哈希内容那同一条 URL 的每次请求都会生成不同 ID去重效果直接归零。我最初就踩过这个坑导致库里半小时内插入了几万条“看起来都一样但其实 ID 都不同”的记录。2.3 埋点方式选型contextmanager 为什么比装饰器更好用Python 里做埋点有三条常见路径装饰器、中间件、上下文管理器。我对装饰器不太感冒的原因是它只能包裹整个函数而爬虫采集函数往往是多阶段的——先请求页面、再解析 HTML、再入库你想单独统计“请求阶段用了多久”和“解析阶段用了多久”装饰器就力不从心了。contextmanager 可以精确包住代码区间。把一个 HTTP 请求用with monitor.request(...)包起来这一行的开始和结束就是计时区间抛异常也能在 finally 里被捕获成功失败都会落库。虽然要多写几行 with 块但换来的是对过程的精细控制。用一段时间之后你会发现这个“多写几行”其实让代码更清晰了因为每个关键动作都被显式标注了监控边界。2.4 存储到底选 SQLite 还是 MySQL很多读者第一反应是“我用 MySQL”。但如果爬虫还处于单机阶段我的建议是先 SQLite。SQLite 是文件型数据库零配置、零服务端对 Python 标准库天然支持开了 WAL 模式之后读写并发能力在单机场景下完全够用。我做过粗略评估普通机械硬盘上每天十万条请求记录写入毫无压力。什么时候该换 MySQL当你的爬虫跑在多台机器上或者单日请求量上了几十万甚至百万SQLite 不管怎么调都有瓶颈。这时候把存储层换成 MySQL 或 PostgreSQL表结构基本不用改无非是把连接串换成pymysql.connect()。更激进的方案是同步到 ClickHouse 做分析但那是后话。前期用 SQLite 跑通模型比什么都实在。3. 从零构建可运行的监控系统理论部分聊完下面进入可以直接抄作业的实现环节。我会把项目结构、表结构、埋点代码、报告生成和告警通知全列出来你照着建一个工程就能跑。3.1 项目结构与前置准备假设你已经有一个正在写的爬虫项目只需要把监控模块作为独立目录塞进去。项目结构我习惯这样放spider_project/ ├── spiders/ # 现有爬虫代码 ├── monitor/ │ ├── __init__.py │ ├── client.py # ReportClient 核心类 │ ├── models.py # 事件模型常量 │ └── report.py # 报告生成脚本 ├── reports/ # 生成的报告输出目录 ├── data/ │ └── spider_reports.db # SQLite 数据库文件 └── config.py # 告警规则等配置依赖非常少Python 3.8 即可requests 用于发告警请求matplotlib 用于画趋势图可选不装也能用 HTML 表格顶上。如果你的爬虫本身已经装了 requests、bs4那这个监控模块几乎不会增加额外负担。3.2 数据库表设计4张表撑起整个监控链路我用 4 张表就把整个链路串起来了runs 表存 run 级事件requests 表存每个请求的明细items 表存采集到的数据指纹alerts 表存告警记录和确认状态。DDL 直接贴在下面。CREATE TABLE IF NOT EXISTS runs ( run_id TEXT PRIMARY KEY, task_id TEXT NOT NULL, name TEXT NOT NULL, status TEXT DEFAULT running, start_time REAL NOT NULL, end_time REAL, duration REAL, seed TEXT, total_requests INTEGER DEFAULT 0, success_requests INTEGER DEFAULT 0, failed_requests INTEGER DEFAULT 0, total_items INTEGER DEFAULT 0, error TEXT ); CREATE TABLE IF NOT EXISTS requests ( request_id TEXT NOT NULL, run_id TEXT NOT NULL, task_id TEXT NOT NULL, url TEXT NOT NULL, method TEXT NOT NULL, status TEXT NOT NULL, duration REAL NOT NULL, error TEXT, request_time REAL NOT NULL, extra TEXT ); CREATE TABLE IF NOT EXISTS items ( item_id INTEGER PRIMARY KEY AUTOINCREMENT, run_id TEXT NOT NULL, task_id TEXT NOT NULL, item_type TEXT NOT NULL, item_key TEXT NOT NULL, content_hash TEXT, created_time REAL NOT NULL, extra TEXT ); CREATE TABLE IF NOT EXISTS alerts ( id INTEGER PRIMARY KEY AUTOINCREMENT, run_id TEXT NOT NULL, alert_type TEXT NOT NULL, message TEXT NOT NULL, created_time REAL NOT NULL, status TEXT DEFAULT pending, sent_time REAL, ack_time REAL, ack_by TEXT );细化一下设计意图。runs 表里的seed字段很重要它记录任务启动的种子 URL 和配置哈希是“可复现”的物理载体。requests 表故意不设主键而是用request_id run_id做业务唯一因为同一个 request_id 可能在同一 run 里出现多次请求比如重试每次请求的表现都值得单独记录。items 表里的content_hash用来快速判断两条数据是否内容一致后续做去重很有帮助。alerts 表的status字段默认为pending确认后改成acknowledged这个机制对应对接传统大平台时“告警如何手动消除”的场景。3.3 埋点模块核心代码埋点模块是整个系统的灵魂我封装了一个ReportClient类内部维护了线程独立的 SQLite 连接并提供两个上下文管理器monitor_run和monitor_request。代码不长但每个细节都值得细看。import sqlite3 import time import uuid import hashlib import json import threading from contextlib import contextmanager def make_request_id(method, url, paramsNone, window300): 生成请求指纹。window秒内同一个请求算同一个ID。 normalized url.split(?)[0] if params: normalized | json.dumps(params, sort_keysTrue) stamp int(time.time() // window) raw f{method}|{normalized}|{stamp} return hashlib.md5(raw.encode()).hexdigest() class ReportClient: def __init__(self, db_pathspider_reports.db): self.db_path db_path self._local threading.local() self._batch [] # 本地批量缓冲 self._batch_lock threading.Lock() self._init_db() def _conn(self): if getattr(self._local, conn, None) is None: conn sqlite3.connect(self.db_path, timeout10) conn.execute(PRAGMA journal_modeWAL;) conn.execute(PRAGMA busy_timeout5000;) self._local.conn conn return self._local.conn def _init_db(self): conn self._conn() conn.executescript( ... 上面那段DDL ... ) conn.commit() def _safe(self, func, *args, **kwargs): 保证监控侧的异常不污染爬虫主流程。 try: return func(*args, **kwargs) except Exception: return None def start_run(self, task_id, name, seed): run_id uuid.uuid4().hex conn self._conn() conn.execute( INSERT INTO runs (run_id, task_id, name, status, start_time, seed) VALUES (?,?,?,?,?,?), (run_id, task_id, name, running, time.time(), seed) ) conn.commit() return run_id def finish_run(self, run_id, status, errorNone): conn self._conn() row conn.execute(SELECT start_time FROM runs WHERE run_id?, (run_id,)).fetchone() if not row: return duration time.time() - row[0] conn.execute( UPDATE runs SET status?, end_time?, duration?, error? WHERE run_id?, (status, time.time(), duration, error, run_id) ) # 统计信息由 report.py 定时聚合这里不重复计算 conn.commit() def record_request(self, run_id, task_id, url, method, status, duration, errorNone, extraNone): # 这里回收集模式要快速返回不能影响爬虫 req_time time.time() request_id make_request_id(method, url, (extra or {}).get(params)) payload (request_id, run_id, task_id, url, method, status, duration, error, req_time, json.dumps(extra, ensure_asciiFalse)) with self._batch_lock: self._batch.append(payload) if len(self._batch) 50: batch list(self._batch) self._batch.clear() if batch: self._safe(self._flush, batch) def _flush(self, batch): conn self._conn() conn.executemany( INSERT INTO requests (request_id, run_id, task_id, url, method, status, duration, error, request_time, extra) VALUES (?,?,?,?,?,?,?,?,?,?), batch ) conn.commit() contextmanager def request(self, run_id, task_id, url, methodGET, extraNone): start time.time() status success error None try: yield except Exception as exc: status failed error repr(exc) raise finally: duration time.time() - start self._safe(self.record_request, run_id, task_id, url, method, status, duration, error, extra) contextmanager def monitor_run(client, task_id, name, seed): run_id client.start_run(task_id, name, seed) status success error None try: yield run_id except Exception as exc: status failed error repr(exc) raise finally: client.finish_run(run_id, status, error)有几个细节特别说明一下。第一_safe把监控侧所有异常都拦住因为埋点代码一旦抛异常会直接影响爬虫本身的稳定性这属于系统设计红线。第二request上下文管理器里 yield 之前不做任何事所有记录都在 finally 中完成所以即使业务代码中途崩了请求记录依然能落库。第三record_request先写进内存批量缓冲累积 50 条再一次性 executemany 写库这是避免“监控写库拖慢爬虫”的关键优化。3.4 运行报告与图表生成数据落库之后报告生成就简单了。我写了一个report.py做三件事聚合 run 级摘要、按任务生成统计表、输出 HTML 和 JSON 报告。import sqlite3 import json import time from collections import Counter def load_runs(db_path, task_idNone): conn sqlite3.connect(db_path) sql SELECT r.run_id, r.task_id, r.name, r.status, r.duration, r.total_requests, r.success_requests, r.failed_requests, r.total_items, r.error FROM runs r if task_id: sql WHERE r.task_id ? rows conn.execute(sql, (task_id,)).fetchall() else: rows conn.execute(sql).fetchall() return rows def summarize_run(db_path, run_id): conn sqlite3.connect(db_path) total conn.execute(SELECT COUNT(1) FROM requests WHERE run_id?, (run_id,)).fetchone()[0] success conn.execute(SELECT COUNT(1) FROM requests WHERE run_id? AND statussuccess, (run_id,)).fetchone()[0] items conn.execute(SELECT COUNT(1) FROM items WHERE run_id?, (run_id,)).fetchone()[0] # 请求耗时分布 durations [r[0] for r in conn.execute( SELECT duration FROM requests WHERE run_id? ORDER BY duration, (run_id,)).fetchall()] avg_dur sum(durations) / len(durations) if durations else 0 p95 durations[int(len(durations) * 0.95)] if len(durations) 20 else (durations[-1] if durations else 0) summary { run_id: run_id, total_requests: total, success_requests: success, failed_requests: total - success, success_rate: round(success / total, 4) if total else 0, avg_duration: round(avg_dur, 3), p95_duration: round(p95, 3), total_items: items, } return summary def generate_html_report(db_path, task_idNone): rows load_runs(db_path, task_id) summaries [] for r in rows: summaries.append({ run_id: r[0], task_id: r[1], name: r[2], status: r[3], duration: r[4], error: r[9], detail: summarize_run(db_path, r[0]) }) html [htmlheadmeta charsetutf-8/headbody] html.append(h1爬虫运行报告/h1) html.append(table border1 cellspacing0 cellpadding6) html.append(trth任务/thth状态/thth耗时(s)/thth总请求/thth成功率/thth平均耗时(s)/ththp95(s)/thth入库条数/thth错误/th/tr) for s in summaries: d s[detail] html.append( ftrtd{s[name]}/tdtd{s[status]}/tdtd{s[duration]}/td ftd{d[total_requests]}/tdtd{d[success_rate]}/tdtd{d[avg_duration]}/td ftd{d[p95_duration]}/tdtd{d[total_items]}/tdtd{s[error] or }/td/tr ) html.append(/table/body/html) return .join(html)图表部分我通常加上一个每日失败率趋势图用 matplotlib 画出来保存成 png在 HTML 报告里引入。画图本身不复杂但要注意中文字体设置否则图里全是方框。Windows 下指定SimHeiLinux 下如果有Noto Sans CJK就用它。这一块可以做成可选依赖不装 matplotlib 也能先用表格顶上。3.5 告警通知与手动确认机制告警是整个监控系统里最容易让人半夜惊醒的部分但也是最容易“狼来了”的部分——发太多没用的告警同事会直接把机器人静音。我的做法是设两条基础规则再加一条稳定性规则。规则一成功率阈值。当一次 run 里请求成功率低于 0.95 时触发。规则二耗时阈值。p95 耗时超过 2 秒触发。规则三连续失败。连续两次以上 run 状态为 failed 才触发避免单次抖动就报警。import requests def send_webhook(webhook_url, message): 发送告警到企业微信/钉钉/飞书群机器人消息结构可自行按平台调整。 payload { msgtype: text, text: {content: message} } resp requests.post(webhook_url, jsonpayload, timeout5) resp.raise_for_status()发送之前先查 alerts 表看同类型告警是否在冷却时间内。我设置了一个 10 分钟冷却窗口同一类型的告警在窗口内不重复发送。确认机制则对应传统大平台的“手动消除告警”把 status 从pending更新为acknowledged同时记录确认人和确认时间。我特意做了一个命令行小工具来执行这个确认动作这样值班的人收到告警后只需一条命令就能标记“已处理”。这点和 Zabbix 里的“告警确认”是同一个思路只是实现更贴合我们自己的场景。3.6 改造一个现有爬虫前后对比说了这么多实际接进爬虫的改动量到底有多大我用一个常见的采集函数来对比一下。改造前def fetch_goods(goods_id): url fhttps://example.com/goods/{goods_id} resp requests.get(url, headersHEADERS) if resp.status_code ! 200: return None data parse_detail(resp.text) save_to_db(data)改造后def fetch_goods(client, run_id, goods_id): url fhttps://example.com/goods/{goods_id} with client.request(run_id, task_idgoods_detail, urlurl, extra{goods_id: goods_id}) as req_id: resp requests.get(url, headersHEADERS) if resp.status_code ! 200: raise ValueError(fhttp {resp.status_code}) data parse_detail(resp.text) save_to_db(data) client.record_item(run_id, task_idgoods_detail, item_typegoods, item_keystr(goods_id))改动就三处函数开头多带一个client和run_id参数请求外层包一个 with 块解析成功后多一行记录 item 的调用。原有业务逻辑一行没动。这就是 contextmanager 方案侵入性低的具体体现不管你的函数是请求详情页、翻列表页还是下载图片包一层 with 就完事。撑几天之后你会习惯所有关键动作都带监控边界。4. 常见问题与排查技巧实录实打实跑这套系统之后我踩过不少坑。下面这六个问题基本覆盖了新手会遇到的大部分雷区每一个我都给了排查思路和解决方案。4.1 埋点拖慢爬虫速度怎么办表现加了埋点之后原来每秒 20 个请求的爬虫掉到每秒 5 个。原因很简单每个请求都实时 INSERT 一次 SQLite磁盘 IO 成了瓶颈。解法批量写入。我最初是每个请求都 commit后来改成内存缓冲 50 条合并 executemany速度快了不止一个量级。另外可以把存储拆到另一个磁盘但大多数场景批量写入就够了。如果并发协程上千单连接批量写也有锁竞争那就把批量缓冲改成监听队列用独立线程消费写库爬虫主线程只往队列塞数据完全不碰磁盘。4.2 database is locked 这个坑怎么填表现爬虫跑着跑着监控模块抛sqlite3.OperationalError: database is locked。原因SQLite 默认在多个连接并发写时可能互相锁库特别是在重试线程多的场景。解法两步走。第一建表后执行PRAGMA journal_modeWAL;让读写并发起来。第二连接时设置timeout10和PRAGMA busy_timeout5000。这两步做完单机并发几十个写线程都没问题。如果还有锁多半是某个长事务没提交检查代码里是不是忘了commit()。4.3 监控代码异常把主流程带崩了表现爬虫没写错但因为监控侧的表结构变更、字段类型不匹配导致业务代码里到处抛异常。这是埋点系统的红线事故。解法所有监控侧方法都包一层 try/except记录调用失败时只打印日志绝不抛出。我写的_safe方法就是干这个的。还有一点值得注意contextmanager 里的 yield 要包得很谨慎不能让监控的异常吃掉业务异常。finally 块里的记录逻辑要用_safe兜底这样就算监控记录失败原始异常照样能往上抛。4.4 任务失败后状态还是 running表现爬虫进程被强杀或者断电runs 表里留下一堆running状态的任务报告看起来像“所有任务都在跑”。解法状态机设计要保证兜底。正常流程里 finish_run 会在 finally 中执行所以只要代码不是os._exit()强退都能更新状态。但强杀场景没法避免我额外加了清扫脚本每次启动报告生成时把超过 30 分钟仍是running状态的 run 统一标记为failed并写入errorKILLED/UNEXPECTED_EXIT。这样报告永远不会有永远在跑的假象。4.5 告警轰炸与去重方案表现某一次网络抖动导致 30% 请求失败告警触发了同时机器人发来 50 条相同消息群里的其他同事直接把你拉黑。解法双保险。一是冷却逻辑同 task 同 alert_type 在 10 分钟内只发一次。二是状态确认机制告警发送后进入pending值班人员处理完改成acknowledged后续同类型告警才能再次触发。冷却窗口的时长可以调但我觉得 10 分钟是个比较合理的起点——既能及时知道问题又不至于半夜一小时收 6 条重复消息。4.6 报告数据对不上时区、口径与采样表现报告里某天的请求数和实际日志差了一大截或者任务耗时忽高忽低。排查顺序先看时区。数据库里存的时间戳我统一用time.time()浮点秒不混用本地时间字符串展示层再按东八区格式化避免时区换算的隐性 bug。再看口径。是“发出请求数”还是“收到响应数”是“成功状态码”还是“解析成功数”这些指标在报告里要明确区分否则前后两天对比完全没意义。最后看采样。requests 表里如果做了批量缓冲最后一批数据可能还没落库就生成了报告我通常在报告脚本开头加一句“先 flush 缓冲、再查库”或者干脆等 30 秒再跑报告。5. 进阶扩展分布式适配与反爬信号观测基础版跑通之后有几个方向值得继续深化。这里根据我自己的实践挑两个对爬虫工程师最实用的扩展来讲。5.1 多进程和多机采集时怎么聚合当你的爬虫从单脚本变成 Scrapy、Celery 多 worker 时埋点数据会分散到多个进程或机器上。最简单的方式是每个进程各写各的 SQLite 文件然后每天统一由报告脚本把多个 db 文件合并成一张总表。合并时按 run_id 维度做 cup 联合把每个文件里的 runs、requests、items 表 UNION 到同一个目标库。注意 requests 表里不要跨 run 做唯一约束否则合并时容易冲突。再往后请求量到了每天几十万、上百万文件合并就不够看了。这时候可以改成所有 worker 把埋点事件发到一个本地 Kafka 或 Redis Stream由单独一个消费进程批量写 ClickHouse。事件模型还是四级字段也还是那些所以迁移成本可控。5.2 把“反爬信号”纳入监控指标很多爬虫工程师会忽略的一点是监控不能只看响应状态码更要看“目标站的拦截信号”。我在 requests 表里设计了一个extra字段专门记录检测到的异常特征。比如请求完成后检测一下响应内容是否包含验证码、滑块、异常访问提示等关键词把这些关键词命中的情况打标存进库。常见反爬手段和对应监控指标可以这样对应反爬手段对应监控指标建议埋点字段返回 403/429被限流请求占比http_status页面出现验证码/滑块验证码触发率captcha_hit返回 JS 挑战页面环境检测命中率js_challenge登录态失效跳登录会话失效次数session_expired返回空数据/占位数据空数据率empty_response有了这些指标你就能在报告里看到“今天成功率 98%但其中有 15% 的请求命中验证码”这种隐藏信息。当验证码命中率连续走高基本可以判断目标站开始针对你的采集模式做防御这时候就该调整请求频率或换采集策略而不是盯着成功率盲目重试。作为爬虫开发者反过来去理解服务端这些防护技术比如 controller 层的限流、前端 JS 挑战对设计监控维度非常有帮助。5.3 用监控数据反推爬虫质量监控系统的价值不只是“出事时救火”更是“平时能指导优化”。我每个月底会翻一下报告里的几个关键趋势平均请求耗时、p95耗时、失败分布、入库有效率。平均耗时持续走高往往是目标站响应变慢或者本机 IP 被限流后请求被反复重试。失败集中在特定 URL 模式可能是某个字段反爬特别严。入库有效率低则是解析代码有 bug比如数据结构调整后筛选条件失配。我建议每个爬虫任务都设定一个“健康基线”。比如正常情况下成功率高于 97%、p95 耗时低于 1.5 秒、入库有效率高于 90%。报告生成时自动和基线对比偏差超过 5% 就黄牌告警超过 10% 就红牌。这比单纯设死阈值更科学因为不同目标站的响应特性差异很大一个固定的“2秒”阈值对有些站是天花板对另一些站却是底线。说到底这套监控系统的最终目的不是“让爬虫看起来跑得很好”而是“让爬虫出问题时你能少死一点脑细胞”。它不会替你解决反爬对抗但能在你被封、被限、代码报错这些糟心事发生时三分钟定位问题在哪一环。我建议你先别急着上重型框架把埋点、存储、报告这条链路跑通比什么都实在。最后再分享一个小技巧。刚开始跑监控时别贪多先把核心指标做扎实请求成功率、耗时分布、入库量。跑两周之后你会发现哪些指标真正影响你的判断再逐步加“反爬命中率”“按 URL 分组的失败热区”这些进阶维度。监控系统是一个越用越知道自己要什么的工具第一版做得粗糙没关系关键是先把观测的口子打开——让你写的每一个爬虫从黑盒变成白盒。这套系统我本人在本地跑了很久稳定的收益就是以后任何一次任务异常我都能在十分钟内说出“哪个环节、哪个请求、哪个时间点出的问题”这个能力对爬虫工程师来说比新版框架值钱得多。 Python爬虫运行监控系统：可复现报告与告警实现
写爬虫的人手机里大概都存着两样东西一份爬虫日志还有一个半夜响起来的报警电话。我见过太多同事凌晨两三点爬起来打开终端盯着日志滚动只为确认采集任务到底卡在哪个环节。数据没抓到还算小事错过窗口期、被目标网站识别成异常流量、又或者因为一次静默小异常导致整个任务失败那才真正让人头疼。今天想聊的这套系统就是专门解决这个问题的——一个可复现的Python爬虫运行报告监控系统。它的核心思路是给爬虫装上“行车记录仪”和“仪表盘”让每一次运行都有迹可循、有数可查、有问题能报警而且整个方案足够轻量不依赖重型运维组件。这套系统更像是一个贴合爬虫场景的轻量级埋点框架加报告中心而不是传统意义上的监控平台。你不用部署 Zabbix、Prometheus 全家桶只需要在爬虫代码的关键节点埋入监控点把请求状态、耗时、结果、异常记录到本地存储再由报告模块自动生成运行报告和告警通知。单机脚本可以接分布式采集集群也可以平移到统一观测。适合的读者很明确写过一段时间爬虫、开始觉得“跑是能跑但跑得怎么样完全靠猜”的人以及需要定期向团队或客户展示采集任务稳定性的人。1. 方案设计可复现和监控到底在解决什么问题先聊聊“可复现”这个词。很多爬虫项目本质上是个黑盒跑完了就是跑完了失败了你也只知道失败了。但监控系统要解决的是“为什么这次失败”“上次成功时候的状态是什么”“同一个请求在前后两次运行中表现有什么差异”。没有这些信息谈优化就是盲人摸象。1.1 可复现的三个层面我把可复现拆成三个层面来理解。第一层是运行过程可复现。每次爬虫任务启动时生成一个全局唯一的 run_id任务里的每个 HTTP 请求再生成一个 request_id。以后无论什么时候翻数据库只要拿到这个 ID就能把整次任务的完整轨迹拉出来——包括请求了哪些 URL、每个请求耗时多少、返回什么状态、是否被反爬拦截、最终入了几条数据。这相当于飞行记录黑匣子。第二层是问题场景可复现。当异常发生时不再是日志里孤零零的一行报错而是带着上下文当时请求的 URL 是什么、携带了哪些参数、从发起请求到抛出异常经历了多少秒、错误堆栈有哪些关键帧。有了这些上下文你才有可能用一个最小脚本把线上异常复原出来然后在本地修修补补而不是靠“听同事描述”来猜问题。第三层是结果可复现。同一套代码、同一个种子 URL、同一份配置前后跑两次结果应该能对齐。如果哪次结果明显异常你能通过报告里的指标快照一眼看出差异点在哪个环节。这有点像做菜时写好菜谱配方不变、火候不变出品就该稳定万一哪天锅糊了你至少能查出是哪一步火大了。1.2 为什么我不推荐直接用 Zabbix 或 Prometheus自己做过运维或者接过公司监控体系的读者可能会有疑问现在开源监控工具这么多直接接一套不是更省事我的观点是传统监控系统解决的是“机器健康”问题而爬虫需要的是“业务过程”观测。Zabbix、Prometheus 这类工具擅长采集 CPU、内存、带宽、进程数这些主机指标也能做告警但它们对“这次爬虫任务里有多少请求被反爬拦截了”“接口响应平均耗时多少、95分位耗时多少”“解析阶段和请求阶段各花了几秒”这类业务级信息整理起来要绕很大一圈。你可能要自定义 exporter、修改告警规则、做一堆额外开发最后得出的报表还不够直白。反过来如果你自己维护一层轻量埋点字段完全由你说了算报告长什么样也由你设计。等爬虫平台真的发展到多机分布式、日请求量百万以上再把同一套事件模型平移到 Kafka ClickHouse也完全顺理成章。前期没必要一上来就背一个重型框架。1.3 整体模块划分从埋点到报告的一整条链路这套监控系统的链路不算复杂拆开看就五个环节采集、存储、报告、告警、可视化。采集层负责在爬虫代码里埋点通过一个上下文管理器把 run 和 request 的执行过程记录成结构化事件。存储层默认使用 SQLite不引入外部服务单机环境下足够稳定。报告层定时或按需从库里聚合数据输出 JSON 摘要、HTML 报告和图表。告警层在发现成功率下跌、耗时超标、任务连续失败时通过企业微信、钉钉或邮件发通知并且在墙上留一个“手动确认”的按钮。可视化层则通过定时刷新报告页面或接入 BI 工具让团队随时能打开看。各层的具体职责可以看这张表模块核心职责落地方式采集层埋点记录 run / request / item 事件Python contextmanager侵入性小存储层保存事件明细与运行状态SQLite WAL 模式后续可换 MySQL报告层聚合指标、生成摘要与图表Python 脚本 matplotlib输出 HTML/JSON告警层阈值判断、发送通知、支持确认自定义规则 Webhook / SMTP可视化层让团队随时查看运行概况定时刷新 HTML 报告或 BI 接入这条链路建好后日常维护爬虫的心态会明显不一样不再是“默默祈祷它不要挂”而是“挂了也知道挂在哪、为什么挂、下次怎么不挂”。2. 核心细节解析埋点模型、运行指纹与存储选址在看代码之前有几个设计上的关键点需要先讲明白。这些点决定了整套系统好不好用、能不能扩展到分布式场景也决定了你在接进现有爬虫时会不会想骂人。2.1 把爬虫运行拆成四级事件模型事件模型的划分直接决定了报告的粒度。我这边的习惯是拆成 run、task、request、item 四级。run 表示一次完整的任务执行比如“2025-06-01 的商品详情页全量采集”它有开始时间、结束时间、最终状态、总请求数、成功数、失败数和入库条数。task 表示一类采集任务比如“商品详情页采集任务”“评论采集任务”方便你按业务归类多个 run。request 表示一次真实的 HTTP 请求记录 URL、请求方法、耗时、状态、错误信息以及反爬信号。item 表示一条解析后成功入库的数据记录数据的主键标识方便对比“请求了 1000 次但最终只有 800 条有效数据”这类偏差。这样拆分的原因很直接日常排查顺序是“哪类任务挂了 - 哪次运行挂了 - 哪批请求丢了 - 哪条数据错了”。四级事件正好支撑这样的逐层钻取将来做分布式也天然适配——不同 worker 采集的事件都带同一个 run_id合并时按 run_id 分区即可。2.2 运行指纹让每次请求都“查得到、可重放”这里的指纹有两类。一个是 run_id负责标识一次任务执行直接用 UUID 就够了不用纠结。另一个是 request_id用来标识一个业务意义上的“重复请求”它的生成要仔细设计。我把 request_id 设计成对请求内容做哈希的结果请求方法 标准化后的 URL 排序后的查询参数 一个时间窗口。时间窗口是关键比如 5 分钟内发起的同一个请求算同一个 ID超过 5 分钟就重新生成。这样设计有两个好处一是天然支持去重重试同一条 URL 不会产生一堆重复记录二是允许一定时间后重新请求因为反爬策略触发后往往需要等一会儿再补采。这里有个细节很坑如果你把精确到秒的当前时间也放进哈希内容那同一条 URL 的每次请求都会生成不同 ID去重效果直接归零。我最初就踩过这个坑导致库里半小时内插入了几万条“看起来都一样但其实 ID 都不同”的记录。2.3 埋点方式选型contextmanager 为什么比装饰器更好用Python 里做埋点有三条常见路径装饰器、中间件、上下文管理器。我对装饰器不太感冒的原因是它只能包裹整个函数而爬虫采集函数往往是多阶段的——先请求页面、再解析 HTML、再入库你想单独统计“请求阶段用了多久”和“解析阶段用了多久”装饰器就力不从心了。contextmanager 可以精确包住代码区间。把一个 HTTP 请求用with monitor.request(...)包起来这一行的开始和结束就是计时区间抛异常也能在 finally 里被捕获成功失败都会落库。虽然要多写几行 with 块但换来的是对过程的精细控制。用一段时间之后你会发现这个“多写几行”其实让代码更清晰了因为每个关键动作都被显式标注了监控边界。2.4 存储到底选 SQLite 还是 MySQL很多读者第一反应是“我用 MySQL”。但如果爬虫还处于单机阶段我的建议是先 SQLite。SQLite 是文件型数据库零配置、零服务端对 Python 标准库天然支持开了 WAL 模式之后读写并发能力在单机场景下完全够用。我做过粗略评估普通机械硬盘上每天十万条请求记录写入毫无压力。什么时候该换 MySQL当你的爬虫跑在多台机器上或者单日请求量上了几十万甚至百万SQLite 不管怎么调都有瓶颈。这时候把存储层换成 MySQL 或 PostgreSQL表结构基本不用改无非是把连接串换成pymysql.connect()。更激进的方案是同步到 ClickHouse 做分析但那是后话。前期用 SQLite 跑通模型比什么都实在。3. 从零构建可运行的监控系统理论部分聊完下面进入可以直接抄作业的实现环节。我会把项目结构、表结构、埋点代码、报告生成和告警通知全列出来你照着建一个工程就能跑。3.1 项目结构与前置准备假设你已经有一个正在写的爬虫项目只需要把监控模块作为独立目录塞进去。项目结构我习惯这样放spider_project/ ├── spiders/ # 现有爬虫代码 ├── monitor/ │ ├── __init__.py │ ├── client.py # ReportClient 核心类 │ ├── models.py # 事件模型常量 │ └── report.py # 报告生成脚本 ├── reports/ # 生成的报告输出目录 ├── data/ │ └── spider_reports.db # SQLite 数据库文件 └── config.py # 告警规则等配置依赖非常少Python 3.8 即可requests 用于发告警请求matplotlib 用于画趋势图可选不装也能用 HTML 表格顶上。如果你的爬虫本身已经装了 requests、bs4那这个监控模块几乎不会增加额外负担。3.2 数据库表设计4张表撑起整个监控链路我用 4 张表就把整个链路串起来了runs 表存 run 级事件requests 表存每个请求的明细items 表存采集到的数据指纹alerts 表存告警记录和确认状态。DDL 直接贴在下面。CREATE TABLE IF NOT EXISTS runs ( run_id TEXT PRIMARY KEY, task_id TEXT NOT NULL, name TEXT NOT NULL, status TEXT DEFAULT running, start_time REAL NOT NULL, end_time REAL, duration REAL, seed TEXT, total_requests INTEGER DEFAULT 0, success_requests INTEGER DEFAULT 0, failed_requests INTEGER DEFAULT 0, total_items INTEGER DEFAULT 0, error TEXT ); CREATE TABLE IF NOT EXISTS requests ( request_id TEXT NOT NULL, run_id TEXT NOT NULL, task_id TEXT NOT NULL, url TEXT NOT NULL, method TEXT NOT NULL, status TEXT NOT NULL, duration REAL NOT NULL, error TEXT, request_time REAL NOT NULL, extra TEXT ); CREATE TABLE IF NOT EXISTS items ( item_id INTEGER PRIMARY KEY AUTOINCREMENT, run_id TEXT NOT NULL, task_id TEXT NOT NULL, item_type TEXT NOT NULL, item_key TEXT NOT NULL, content_hash TEXT, created_time REAL NOT NULL, extra TEXT ); CREATE TABLE IF NOT EXISTS alerts ( id INTEGER PRIMARY KEY AUTOINCREMENT, run_id TEXT NOT NULL, alert_type TEXT NOT NULL, message TEXT NOT NULL, created_time REAL NOT NULL, status TEXT DEFAULT pending, sent_time REAL, ack_time REAL, ack_by TEXT );细化一下设计意图。runs 表里的seed字段很重要它记录任务启动的种子 URL 和配置哈希是“可复现”的物理载体。requests 表故意不设主键而是用request_id run_id做业务唯一因为同一个 request_id 可能在同一 run 里出现多次请求比如重试每次请求的表现都值得单独记录。items 表里的content_hash用来快速判断两条数据是否内容一致后续做去重很有帮助。alerts 表的status字段默认为pending确认后改成acknowledged这个机制对应对接传统大平台时“告警如何手动消除”的场景。3.3 埋点模块核心代码埋点模块是整个系统的灵魂我封装了一个ReportClient类内部维护了线程独立的 SQLite 连接并提供两个上下文管理器monitor_run和monitor_request。代码不长但每个细节都值得细看。import sqlite3 import time import uuid import hashlib import json import threading from contextlib import contextmanager def make_request_id(method, url, paramsNone, window300): 生成请求指纹。window秒内同一个请求算同一个ID。 normalized url.split(?)[0] if params: normalized | json.dumps(params, sort_keysTrue) stamp int(time.time() // window) raw f{method}|{normalized}|{stamp} return hashlib.md5(raw.encode()).hexdigest() class ReportClient: def __init__(self, db_pathspider_reports.db): self.db_path db_path self._local threading.local() self._batch [] # 本地批量缓冲 self._batch_lock threading.Lock() self._init_db() def _conn(self): if getattr(self._local, conn, None) is None: conn sqlite3.connect(self.db_path, timeout10) conn.execute(PRAGMA journal_modeWAL;) conn.execute(PRAGMA busy_timeout5000;) self._local.conn conn return self._local.conn def _init_db(self): conn self._conn() conn.executescript( ... 上面那段DDL ... ) conn.commit() def _safe(self, func, *args, **kwargs): 保证监控侧的异常不污染爬虫主流程。 try: return func(*args, **kwargs) except Exception: return None def start_run(self, task_id, name, seed): run_id uuid.uuid4().hex conn self._conn() conn.execute( INSERT INTO runs (run_id, task_id, name, status, start_time, seed) VALUES (?,?,?,?,?,?), (run_id, task_id, name, running, time.time(), seed) ) conn.commit() return run_id def finish_run(self, run_id, status, errorNone): conn self._conn() row conn.execute(SELECT start_time FROM runs WHERE run_id?, (run_id,)).fetchone() if not row: return duration time.time() - row[0] conn.execute( UPDATE runs SET status?, end_time?, duration?, error? WHERE run_id?, (status, time.time(), duration, error, run_id) ) # 统计信息由 report.py 定时聚合这里不重复计算 conn.commit() def record_request(self, run_id, task_id, url, method, status, duration, errorNone, extraNone): # 这里回收集模式要快速返回不能影响爬虫 req_time time.time() request_id make_request_id(method, url, (extra or {}).get(params)) payload (request_id, run_id, task_id, url, method, status, duration, error, req_time, json.dumps(extra, ensure_asciiFalse)) with self._batch_lock: self._batch.append(payload) if len(self._batch) 50: batch list(self._batch) self._batch.clear() if batch: self._safe(self._flush, batch) def _flush(self, batch): conn self._conn() conn.executemany( INSERT INTO requests (request_id, run_id, task_id, url, method, status, duration, error, request_time, extra) VALUES (?,?,?,?,?,?,?,?,?,?), batch ) conn.commit() contextmanager def request(self, run_id, task_id, url, methodGET, extraNone): start time.time() status success error None try: yield except Exception as exc: status failed error repr(exc) raise finally: duration time.time() - start self._safe(self.record_request, run_id, task_id, url, method, status, duration, error, extra) contextmanager def monitor_run(client, task_id, name, seed): run_id client.start_run(task_id, name, seed) status success error None try: yield run_id except Exception as exc: status failed error repr(exc) raise finally: client.finish_run(run_id, status, error)有几个细节特别说明一下。第一_safe把监控侧所有异常都拦住因为埋点代码一旦抛异常会直接影响爬虫本身的稳定性这属于系统设计红线。第二request上下文管理器里 yield 之前不做任何事所有记录都在 finally 中完成所以即使业务代码中途崩了请求记录依然能落库。第三record_request先写进内存批量缓冲累积 50 条再一次性 executemany 写库这是避免“监控写库拖慢爬虫”的关键优化。3.4 运行报告与图表生成数据落库之后报告生成就简单了。我写了一个report.py做三件事聚合 run 级摘要、按任务生成统计表、输出 HTML 和 JSON 报告。import sqlite3 import json import time from collections import Counter def load_runs(db_path, task_idNone): conn sqlite3.connect(db_path) sql SELECT r.run_id, r.task_id, r.name, r.status, r.duration, r.total_requests, r.success_requests, r.failed_requests, r.total_items, r.error FROM runs r if task_id: sql WHERE r.task_id ? rows conn.execute(sql, (task_id,)).fetchall() else: rows conn.execute(sql).fetchall() return rows def summarize_run(db_path, run_id): conn sqlite3.connect(db_path) total conn.execute(SELECT COUNT(1) FROM requests WHERE run_id?, (run_id,)).fetchone()[0] success conn.execute(SELECT COUNT(1) FROM requests WHERE run_id? AND statussuccess, (run_id,)).fetchone()[0] items conn.execute(SELECT COUNT(1) FROM items WHERE run_id?, (run_id,)).fetchone()[0] # 请求耗时分布 durations [r[0] for r in conn.execute( SELECT duration FROM requests WHERE run_id? ORDER BY duration, (run_id,)).fetchall()] avg_dur sum(durations) / len(durations) if durations else 0 p95 durations[int(len(durations) * 0.95)] if len(durations) 20 else (durations[-1] if durations else 0) summary { run_id: run_id, total_requests: total, success_requests: success, failed_requests: total - success, success_rate: round(success / total, 4) if total else 0, avg_duration: round(avg_dur, 3), p95_duration: round(p95, 3), total_items: items, } return summary def generate_html_report(db_path, task_idNone): rows load_runs(db_path, task_id) summaries [] for r in rows: summaries.append({ run_id: r[0], task_id: r[1], name: r[2], status: r[3], duration: r[4], error: r[9], detail: summarize_run(db_path, r[0]) }) html [htmlheadmeta charsetutf-8/headbody] html.append(h1爬虫运行报告/h1) html.append(table border1 cellspacing0 cellpadding6) html.append(trth任务/thth状态/thth耗时(s)/thth总请求/thth成功率/thth平均耗时(s)/ththp95(s)/thth入库条数/thth错误/th/tr) for s in summaries: d s[detail] html.append( ftrtd{s[name]}/tdtd{s[status]}/tdtd{s[duration]}/td ftd{d[total_requests]}/tdtd{d[success_rate]}/tdtd{d[avg_duration]}/td ftd{d[p95_duration]}/tdtd{d[total_items]}/tdtd{s[error] or }/td/tr ) html.append(/table/body/html) return .join(html)图表部分我通常加上一个每日失败率趋势图用 matplotlib 画出来保存成 png在 HTML 报告里引入。画图本身不复杂但要注意中文字体设置否则图里全是方框。Windows 下指定SimHeiLinux 下如果有Noto Sans CJK就用它。这一块可以做成可选依赖不装 matplotlib 也能先用表格顶上。3.5 告警通知与手动确认机制告警是整个监控系统里最容易让人半夜惊醒的部分但也是最容易“狼来了”的部分——发太多没用的告警同事会直接把机器人静音。我的做法是设两条基础规则再加一条稳定性规则。规则一成功率阈值。当一次 run 里请求成功率低于 0.95 时触发。规则二耗时阈值。p95 耗时超过 2 秒触发。规则三连续失败。连续两次以上 run 状态为 failed 才触发避免单次抖动就报警。import requests def send_webhook(webhook_url, message): 发送告警到企业微信/钉钉/飞书群机器人消息结构可自行按平台调整。 payload { msgtype: text, text: {content: message} } resp requests.post(webhook_url, jsonpayload, timeout5) resp.raise_for_status()发送之前先查 alerts 表看同类型告警是否在冷却时间内。我设置了一个 10 分钟冷却窗口同一类型的告警在窗口内不重复发送。确认机制则对应传统大平台的“手动消除告警”把 status 从pending更新为acknowledged同时记录确认人和确认时间。我特意做了一个命令行小工具来执行这个确认动作这样值班的人收到告警后只需一条命令就能标记“已处理”。这点和 Zabbix 里的“告警确认”是同一个思路只是实现更贴合我们自己的场景。3.6 改造一个现有爬虫前后对比说了这么多实际接进爬虫的改动量到底有多大我用一个常见的采集函数来对比一下。改造前def fetch_goods(goods_id): url fhttps://example.com/goods/{goods_id} resp requests.get(url, headersHEADERS) if resp.status_code ! 200: return None data parse_detail(resp.text) save_to_db(data)改造后def fetch_goods(client, run_id, goods_id): url fhttps://example.com/goods/{goods_id} with client.request(run_id, task_idgoods_detail, urlurl, extra{goods_id: goods_id}) as req_id: resp requests.get(url, headersHEADERS) if resp.status_code ! 200: raise ValueError(fhttp {resp.status_code}) data parse_detail(resp.text) save_to_db(data) client.record_item(run_id, task_idgoods_detail, item_typegoods, item_keystr(goods_id))改动就三处函数开头多带一个client和run_id参数请求外层包一个 with 块解析成功后多一行记录 item 的调用。原有业务逻辑一行没动。这就是 contextmanager 方案侵入性低的具体体现不管你的函数是请求详情页、翻列表页还是下载图片包一层 with 就完事。撑几天之后你会习惯所有关键动作都带监控边界。4. 常见问题与排查技巧实录实打实跑这套系统之后我踩过不少坑。下面这六个问题基本覆盖了新手会遇到的大部分雷区每一个我都给了排查思路和解决方案。4.1 埋点拖慢爬虫速度怎么办表现加了埋点之后原来每秒 20 个请求的爬虫掉到每秒 5 个。原因很简单每个请求都实时 INSERT 一次 SQLite磁盘 IO 成了瓶颈。解法批量写入。我最初是每个请求都 commit后来改成内存缓冲 50 条合并 executemany速度快了不止一个量级。另外可以把存储拆到另一个磁盘但大多数场景批量写入就够了。如果并发协程上千单连接批量写也有锁竞争那就把批量缓冲改成监听队列用独立线程消费写库爬虫主线程只往队列塞数据完全不碰磁盘。4.2 database is locked 这个坑怎么填表现爬虫跑着跑着监控模块抛sqlite3.OperationalError: database is locked。原因SQLite 默认在多个连接并发写时可能互相锁库特别是在重试线程多的场景。解法两步走。第一建表后执行PRAGMA journal_modeWAL;让读写并发起来。第二连接时设置timeout10和PRAGMA busy_timeout5000。这两步做完单机并发几十个写线程都没问题。如果还有锁多半是某个长事务没提交检查代码里是不是忘了commit()。4.3 监控代码异常把主流程带崩了表现爬虫没写错但因为监控侧的表结构变更、字段类型不匹配导致业务代码里到处抛异常。这是埋点系统的红线事故。解法所有监控侧方法都包一层 try/except记录调用失败时只打印日志绝不抛出。我写的_safe方法就是干这个的。还有一点值得注意contextmanager 里的 yield 要包得很谨慎不能让监控的异常吃掉业务异常。finally 块里的记录逻辑要用_safe兜底这样就算监控记录失败原始异常照样能往上抛。4.4 任务失败后状态还是 running表现爬虫进程被强杀或者断电runs 表里留下一堆running状态的任务报告看起来像“所有任务都在跑”。解法状态机设计要保证兜底。正常流程里 finish_run 会在 finally 中执行所以只要代码不是os._exit()强退都能更新状态。但强杀场景没法避免我额外加了清扫脚本每次启动报告生成时把超过 30 分钟仍是running状态的 run 统一标记为failed并写入errorKILLED/UNEXPECTED_EXIT。这样报告永远不会有永远在跑的假象。4.5 告警轰炸与去重方案表现某一次网络抖动导致 30% 请求失败告警触发了同时机器人发来 50 条相同消息群里的其他同事直接把你拉黑。解法双保险。一是冷却逻辑同 task 同 alert_type 在 10 分钟内只发一次。二是状态确认机制告警发送后进入pending值班人员处理完改成acknowledged后续同类型告警才能再次触发。冷却窗口的时长可以调但我觉得 10 分钟是个比较合理的起点——既能及时知道问题又不至于半夜一小时收 6 条重复消息。4.6 报告数据对不上时区、口径与采样表现报告里某天的请求数和实际日志差了一大截或者任务耗时忽高忽低。排查顺序先看时区。数据库里存的时间戳我统一用time.time()浮点秒不混用本地时间字符串展示层再按东八区格式化避免时区换算的隐性 bug。再看口径。是“发出请求数”还是“收到响应数”是“成功状态码”还是“解析成功数”这些指标在报告里要明确区分否则前后两天对比完全没意义。最后看采样。requests 表里如果做了批量缓冲最后一批数据可能还没落库就生成了报告我通常在报告脚本开头加一句“先 flush 缓冲、再查库”或者干脆等 30 秒再跑报告。5. 进阶扩展分布式适配与反爬信号观测基础版跑通之后有几个方向值得继续深化。这里根据我自己的实践挑两个对爬虫工程师最实用的扩展来讲。5.1 多进程和多机采集时怎么聚合当你的爬虫从单脚本变成 Scrapy、Celery 多 worker 时埋点数据会分散到多个进程或机器上。最简单的方式是每个进程各写各的 SQLite 文件然后每天统一由报告脚本把多个 db 文件合并成一张总表。合并时按 run_id 维度做 cup 联合把每个文件里的 runs、requests、items 表 UNION 到同一个目标库。注意 requests 表里不要跨 run 做唯一约束否则合并时容易冲突。再往后请求量到了每天几十万、上百万文件合并就不够看了。这时候可以改成所有 worker 把埋点事件发到一个本地 Kafka 或 Redis Stream由单独一个消费进程批量写 ClickHouse。事件模型还是四级字段也还是那些所以迁移成本可控。5.2 把“反爬信号”纳入监控指标很多爬虫工程师会忽略的一点是监控不能只看响应状态码更要看“目标站的拦截信号”。我在 requests 表里设计了一个extra字段专门记录检测到的异常特征。比如请求完成后检测一下响应内容是否包含验证码、滑块、异常访问提示等关键词把这些关键词命中的情况打标存进库。常见反爬手段和对应监控指标可以这样对应反爬手段对应监控指标建议埋点字段返回 403/429被限流请求占比http_status页面出现验证码/滑块验证码触发率captcha_hit返回 JS 挑战页面环境检测命中率js_challenge登录态失效跳登录会话失效次数session_expired返回空数据/占位数据空数据率empty_response有了这些指标你就能在报告里看到“今天成功率 98%但其中有 15% 的请求命中验证码”这种隐藏信息。当验证码命中率连续走高基本可以判断目标站开始针对你的采集模式做防御这时候就该调整请求频率或换采集策略而不是盯着成功率盲目重试。作为爬虫开发者反过来去理解服务端这些防护技术比如 controller 层的限流、前端 JS 挑战对设计监控维度非常有帮助。5.3 用监控数据反推爬虫质量监控系统的价值不只是“出事时救火”更是“平时能指导优化”。我每个月底会翻一下报告里的几个关键趋势平均请求耗时、p95耗时、失败分布、入库有效率。平均耗时持续走高往往是目标站响应变慢或者本机 IP 被限流后请求被反复重试。失败集中在特定 URL 模式可能是某个字段反爬特别严。入库有效率低则是解析代码有 bug比如数据结构调整后筛选条件失配。我建议每个爬虫任务都设定一个“健康基线”。比如正常情况下成功率高于 97%、p95 耗时低于 1.5 秒、入库有效率高于 90%。报告生成时自动和基线对比偏差超过 5% 就黄牌告警超过 10% 就红牌。这比单纯设死阈值更科学因为不同目标站的响应特性差异很大一个固定的“2秒”阈值对有些站是天花板对另一些站却是底线。说到底这套监控系统的最终目的不是“让爬虫看起来跑得很好”而是“让爬虫出问题时你能少死一点脑细胞”。它不会替你解决反爬对抗但能在你被封、被限、代码报错这些糟心事发生时三分钟定位问题在哪一环。我建议你先别急着上重型框架把埋点、存储、报告这条链路跑通比什么都实在。最后再分享一个小技巧。刚开始跑监控时别贪多先把核心指标做扎实请求成功率、耗时分布、入库量。跑两周之后你会发现哪些指标真正影响你的判断再逐步加“反爬命中率”“按 URL 分组的失败热区”这些进阶维度。监控系统是一个越用越知道自己要什么的工具第一版做得粗糙没关系关键是先把观测的口子打开——让你写的每一个爬虫从黑盒变成白盒。这套系统我本人在本地跑了很久稳定的收益就是以后任何一次任务异常我都能在十分钟内说出“哪个环节、哪个请求、哪个时间点出的问题”这个能力对爬虫工程师来说比新版框架值钱得多。