1. 爬虫为什么必须有一套监控告警体系先讲个真实场景。你写了一个爬虫每天凌晨两点定时去抓竞品价格数据平时跑得好好的突然某天对方网站改版了页面结构你的解析规则全部失效爬虫开始疯狂报错或者更糟——静默抓回一堆空数据。如果这个爬虫没有监控你可能要到第二天早上打开数据库看到连续五六个小时的数据全是空的才意识到出事了。更麻烦的是这种问题往往不是一次性的网站改版、IP被限制、验证码升级、接口限流每一个都能让你的爬虫悄悄病死。我这些年维护过的爬虫项目几乎每个都经历过这种半夜翻车的时刻。爬虫本身不难写难的是让它在无人值守的情况下持续稳定运行。所以我才说爬虫监控告警不是可选项是必选项。这篇文章要聊的就是怎么用企业微信或钉钉的机器人搭配一套轻量级的监控逻辑给爬虫装上 7×24 小时的生命体征监测仪。这套系统适合谁适合手里有爬虫任务在跑的个人开发者、运维工程师、数据团队。不管你的爬虫是跑在云服务器上还是跑在公司内网机器上只要你能执行 Python 脚本、能访问外网 API这套方案就能直接落地。它解决的核心问题有三个第一时间知道爬虫挂了、第一时间知道数据异常了、不用半夜爬起来看日志。2. 爬虫监控的整体架构与关键指标设计2.1 监控数据从哪来爬虫的四个关键信号想监控爬虫先得搞清楚监控什么。我见过不少人一上来就盯着进程还在不在其实进程活着不代表爬虫正常进程死了也不一定需要立刻报警——得看场景。我建议把爬虫的监控指标分成四层第一层是进程级。进程是否存活、有没有被系统 OOM Killer 干掉、主循环是否卡死。这一层最简单但只覆盖爬虫跑没跑的问题。第二层是任务级。单次任务是否成功完成、耗时是否异常、有没有重试、最终状态是成功还是失败。这一层解决的是这一轮抓取任务是否正常结束。第三层是数据级。这是最容易忽略、却最重要的一层。抓回来的数据量是否在正常区间新增记录数是零还是突然暴涨去重率是不是异常字段完整性怎么样有时候爬虫明明成功了但因为页面结构变化解析出来的全是空字段任务状态显示成功数据却是一堆垃圾。没有数据级的监控这个坑你根本发现不了。第四层是资源级。CPU、内存、磁盘、带宽。爬虫是资源消耗大户尤其是跑在共享服务器上的爬虫容易被其他任务挤死或者因为日志膨胀把磁盘写满。我自己的实践中这四层指标会映射成几条具体的告警规则每一条都能对应到一个可执行的检查动作。比如进程级用 supervisor 或 systemd 来守护和检查任务级在爬虫入口和出口埋点上报数据级对比历史均值做波动检测资源级直接采集系统指标。2.2 告警规则与阈值怎么定阈值定得太松等于没装监控定得太紧每天告警刷屏几天后你就开始无视所有告警。这里分享几个我踩过坑之后的经验。第一条规则是不要用固定绝对值要用滑动基线。比如单次任务抓取条数低于 100 就告警这个规则看起来很合理但如果你的爬虫抓的是一个流量波动很大的网站周末数据量本来就低这个阈值就会在周末疯狂误报。更好的做法是取最近 7 天同一时段的均值低于均值的 30% 才告警。实现上也不难从数据库里查一下最近七天的记录做个 AVG 就行。第二条规则是连续失败才告警。单次失败可能是网络抖动、目标网站临时抽风不值得把人从被窝里叫起来。我常用的策略是连续 3 次失败或者10 分钟内失败超过 2 次才触发告警。用代码实现就是在内存里维护一个失败计数器重置条件有两个成功一次清零或者超过一段时间自动衰减。第三条规则是区分告警级别。我把告警分成三级WARN 级只发消息不打扰比如单次任务失败但重试成功ERROR 级需要尽快处理比如连续失败、数据量为零CRITICAL 级必须立刻处理比如进程挂了、磁盘快满了。不同级别可以走不同的推送策略低级别合并发送高级别单独秒推。2.3 告警降噪避免凌晨三点被吵醒告警降噪这个词听起来高大上其实核心就一句话让每条告警都有足够的信息量并且避免重复轰炸。最常见的降噪手段是去重聚合。同一个任务在 10 分钟内连续告警 5 次正常人只需要看到一条消息任务 X 连续失败 5 次附带最近一次的错误信息就够了。实现方式是在告警模块里维护一个字典key 是任务名加错误类型value 是首次触发时间和累计次数。推送时如果发现同样的 key 在聚合窗口内已经推过就不再推送只更新计数器。等到窗口结束再补推一条聚合结果。另一个手段是静默窗口。有些告警你知道了也没法立刻处理比如目标网站的反爬策略变化你得等到上班才能分析。这种情况下可以给告警规则配置一个工作时间外只记录不推送的选项或者把夜间告警统一合并成早上的夜间告警汇总。还有一个容易被忽略的降噪点恢复通知。故障恢复之后必须给运维人员发一条已恢复的消息否则大家不知道问题解决了可能白忙一场。恢复通知也要做去重——只有之前推送过故障告警的任务恢复时才推送。3. 告警推送通道的搭建实操3.1 企业微信机器人从创建群聊到拿到 Webhook企业微信机器人是目前我觉得最省事的告警通道。它不需要单独安装客户端也不需要申请什么应用权限只要你能建一个企业微信群就能创建一个机器人拿到一个 Webhook 地址。具体步骤很简单先在企业微信里创建一个群可以只有你自己一个人然后在群设置里找到群机器人点击添加机器人给它起个名字系统会生成一个 Webhook 地址形如https://qyapi.weixin.qq.com/cgi-bin/webhook/send?keyxxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx。拿到 Webhook 之后推送消息只需要一个 HTTP POST 请求。企业微信机器人支持文本text、Markdownmarkdown、图片image、图文news、文件file五种消息类型。对告警场景来说最常用的是 text 和 markdown 两种。有一个很实用的细节markdown 消息的正文会被包裹在一个markdown字段里而且它的语法支持有限不支持 HTML。我一般会把告警信息格式化成文本加简单 Markdown 的混合体比如用**加粗**标注任务名、用引用错误信息这样的阅读体验比纯文本好很多。另外企业微信机器人的消息内容默认不会自动换行需要显式在内容里加\n换行这个坑我第一次踩的时候折腾了十分钟。推送的时候只需要用 requests 发 POST 请求import requests import json def send_wecom(webhook: str, content: str, msg_type: str text): if msg_type text: payload { msgtype: text, text: {content: content} } else: payload { msgtype: markdown, markdown: {content: content} } resp requests.post(webhook, jsonpayload, timeout10) result resp.json() if result.get(errcode) ! 0: raise RuntimeError(f企业微信推送失败: {result.get(errmsg)}) return result这里有几个细节。第一是超时时间必须设置不然企业微信接口万一 hang 住你的告警线程也跟着卡死。第二是返回值里的errcode只有等于 0 才是成功其他值都有具体含义比如93000表示 webhook 不存在或已被删除93001表示机器人已被移出群聊。把这些错误码记录下来排查问题的时候会省很多时间。3.2 钉钉机器人安全设置与加签方式钉钉机器人的整体流程和企业微信类似也是在群里添加自定义机器人然后拿到 Webhook 地址。不过钉钉多了一个安全设置环节有三种方式自定义关键词、加签、IP 白名单。我强烈建议用加签方式因为它不需要额外维护 IP 列表安全性也足够。加签的原理是你设置一个密钥Secret推送时用当前时间戳加上密钥做 HMAC-SHA256 签名然后把签名值拼到 Webhook URL 后面。钉钉服务器会校验这个签名校验通过才会接受消息。签名算法固定是import time import hmac import hashlib import base64 import urllib.parse def dingtalk_sign(secret: str) - str: timestamp str(round(time.time() * 1000)) secret_enc secret.encode(utf-8) string_to_sign f{timestamp}\n{secret}.encode(utf-8) hmac_code hmac.new(secret_enc, string_to_sign, digestmodhashlib.sha256).digest() sign urllib.parse.quote_plus(base64.b64encode(hmac_code)) return timestamp, sign推送时把 Webhook 地址、timestampxxxsignxxx拼在一起再 POST 数据。钉钉的 payload 结构和企业微信略有不同文本消息用 text/markdown 包一层def send_dingtalk(webhook: str, secret: str, content: str, msg_type: str text): timestamp, sign dingtalk_sign(secret) url f{webhook}timestamp{timestamp}sign{sign} if msg_type text: payload { msgtype: text, text: {content: content} } else: payload { msgtype: markdown, markdown: {title: 爬虫告警, text: content} } resp requests.post(url, jsonpayload, timeout10) result resp.json() if result.get(errcode) ! 0: raise RuntimeError(f钉钉推送失败: {result.get(errmsg)}) return result有个容易踩的坑钉钉对消息体的限制比企业微信严格文本消息最大支持 5000 字节Markdown 消息有限制图片消息对 URL 有域名白名单要求。我之前就遇到过因为告警信息里塞了太长的错误堆栈导致推送一直失败后来才知道是被长度限制卡住了。处理办法是在推送前对消息做截断保留前 1500 个字符超出的部分用...完整日志见 server:/path/to/log代替。还有一个细节钉钉机器人每秒只能发送 20 条消息虽然正常情况根本达不到这个量级但如果你从批量告警变成批量轰炸就会被限流。所以告警推送模块里最好加一个简单的并发控制用threading.Semaphore或者直接串行推送即可。3.3 Python 封装统一的告警推送模块企业微信和钉钉各自封装一套代码是可以的但如果你以后想切换或者同时用两个渠道维护成本就上来了。我更推荐的做法是做一个统一的告警客户端接口内部再分发到具体渠道。这样业务代码里只需要调用alert_client.send(task_failed, task_name, error_msg)完全不用关心底层走的是企业微信还是钉钉。class AlertClient: def __init__(self, wecom_webhookNone, dingtalk_webhookNone, dingtalk_secretNone): self.wecom_webhook wecom_webhook self.dingtalk_webhook dingtalk_webhook self.dingtalk_secret dingtalk_secret self._recent_alerts {} # (task_name, alert_type) - last_sent_time def send(self, task_name, alert_type, message, levelERROR): key (task_name, alert_type) now time.time() last_time self._recent_alerts.get(key, 0) if now - last_time 300: # 五分钟内同一类型不重复推送 return False content self._format_message(task_name, alert_type, message, level) errors [] if self.wecom_webhook: try: send_wecom(self.wecom_webhook, content, msg_typemarkdown) except Exception as e: errors.append(f企业微信: {e}) if self.dingtalk_webhook: try: send_dingtalk(self.dingtalk_webhook, self.dingtalk_secret, content, msg_typemarkdown) except Exception as e: errors.append(f钉钉: {e}) if errors: # 两个渠道都失败时写入本地日志避免告警丢失 log_error(f告警推送失败: {errors}) return False self._recent_alerts[key] now return True这个模块虽然简单但把去重、多渠道分发、失败兜底这几件事都做了。尤其是两个渠道都失败的情况我建议一定要在本地留一份日志否则告警本身就丢了你连告警发送失败这件事都不知道那才是真正的灾难。4. 存储、调度与完整链路串起来4.1 用 SQLAlchemy 记录爬虫运行日志与告警历史告警消息推出去就完了吗不是的。我强烈建议把每次爬虫任务的运行状态、每次告警的触发记录都存到数据库里。有两个目的一是做数据级的基线计算前面提到的滑动均值二是事后回溯问题时你能查到这个任务在什么时间点开始失败、失败了多少次、每次的错误信息是什么。存储方案我用的是 SQLAlchemy它作为一个 ORM 层可以无缝切换 SQLite、MySQL、PostgreSQL。对于爬虫监控这种场景量级通常不大SQLite 单文件就能跑但如果你已经有 MySQL 了直接连上也不麻烦。先定义两个核心模型from datetime import datetime from sqlalchemy import create_engine, Column, Integer, String, DateTime, Text, Float from sqlalchemy.orm import declarative_base, sessionmaker Base declarative_base() class CrawlTaskLog(Base): __tablename__ crawl_task_log id Column(Integer, primary_keyTrue, autoincrementTrue) task_name Column(String(128), nullableFalse, indexTrue) status Column(String(16), nullableFalse) # success / failed / timeout started_at Column(DateTime, nullableFalse) finished_at Column(DateTime) item_count Column(Integer, default0) # 抓取条数 error_msg Column(Text) # 失败时的错误信息 duration_sec Column(Float) # 任务耗时 class AlertRecord(Base): __tablename__ alert_record id Column(Integer, primary_keyTrue, autoincrementTrue) task_name Column(String(128), nullableFalse) alert_type Column(String(32), nullableFalse) # task_failed / data_empty / process_down level Column(String(8), nullableFalse) # WARN / ERROR / CRITICAL content Column(Text) created_at Column(DateTime, nullableFalse, defaultdatetime.now)写入逻辑挂在爬虫任务的 finally 块里无论成功失败都记录一条。有个细节要注意如果爬虫本身崩得很惨连 finally 都没执行到怎么办这就需要在爬虫外层再用一个守护进程来兜底后面讲调度的时候会提到。基线计算就是基于CrawlTaskLog表的历史数据做的。比如判断这次抓取条数是否异常偏低就查最近 7 天同任务名的成功记录算平均值和标准差如果本次值低于均值减两倍标准差就认为异常。这个逻辑用 SQLAlchemy 写起来很清晰from sqlalchemy import select, func from datetime import timedelta def check_item_count_anomaly(session, task_name, item_count, days7): since datetime.now() - timedelta(daysdays) result session.execute( select(func.avg(CrawlTaskLog.item_count), func.stddev(CrawlTaskLog.item_count)) .where(CrawlTaskLog.task_name task_name) .where(CrawlTaskLog.status success) .where(CrawlTaskLog.started_at since) ).one() avg, std result if avg is None or item_count avg - 2 * std: return True return False注意stddev函数在 SQLite 里是原生支持的在 MySQL 里叫STDDEVPostgreSQL 里叫STDDEVSQLAlchemy 会帮你翻译不用操心。但如果样本太少比如历史上就两三次成功记录算出来的标准差可能失真所以我在代码里会加一个if count 5: return False的保护逻辑样本太少时不告警避免误报。4.2 定时调度巡检crontab 与 DolphinScheduler 两种方案告警系统本身需要有一个调度机制来驱动。这里分两个层面爬虫任务的定时执行以及监控巡检的定时执行。最简单的方案是 crontab。比如你的爬虫每天凌晨 2 点跑一次就在 crontab 里写0 2 * * * cd /opt/spider /usr/bin/python3 run_task.py logs/task.log 21然后在run_task.py里埋好各种状态上报逻辑。这种方式对单个或少量爬虫足够用但 crontab 有个缺点没有主动的重试机制爬虫进程因为环境问题挂掉之后下一次执行还得等下一个周期。进阶一点的方案是用 DolphinScheduler 这类工作流调度平台。它自带失败重试、超时告警、依赖管理而且可以直接配置任务失败后发起告警的钩子。DolphinScheduler 里创建一个 Shell 节点来跑爬虫命令然后在失败策略里选告警在告警组里配置企业微信或钉钉的 Webhook调度平台本身就把任务失败→发告警这条链路打通了。如果你的爬虫任务比较多、执行流复杂我建议直接用 DolphinScheduler它能省掉你自己写调度和重试逻辑的时间。不过调度平台只管任务是否按计划执行它不管任务执行了但数据异常。所以无论用哪种调度方案我都建议另外跑一个独立巡检脚本每 10 分钟检查一次CrawlTaskLog表如果某个任务本该在特定时间执行完却查不到对应记录说明任务根本没启动这时立即告警。这个巡检脚本本身也放在 crontab 或 systemd timer 里形成一主一备的双重保障。4.3 消息格式设计让告警一眼看懂告警消息的格式直接决定了收到消息的人能不能在 10 秒内判断出问题的严重性。我见过最糟糕的消息是只有一行: 时间连哪个任务、什么错误都不知道。我自己的格式化模板是固定的经过多次迭代后现在的版本长这样**[CRITICAL] 爬虫任务连续失败** - 任务名称product_price_crawler - 失败次数连续 3 次 (时间窗口: 14:00-14:30) - 最新错误AttributeError: NoneType object has no attribute find_all - 最近成功2025-01-05 14:00:00 - 影响范围products 表昨日价格字段缺失 - 处理建议检查目标站页面结构是否变更这个格式有几个设计要点第一行必须有级别和任务名让人一眼定位错误信息必须是最新的不是最早的那条最近成功时间让运维判断问题的持续时间影响范围让运维判断要不要立即处理还是可以等上班再说。实现上任务名和错误信息由爬虫传进来影响范围和处理建议可以由一个简单的规则映射生成。比如提前维护一个任务清单TASK_META { product_price_crawler: { name: 商品价格抓取, impact: products 表价格字段缺失影响比价功能, suggestion: 检查目标站结构 确认解析规则, }, }有了这个映射告警内容就从技术日志变成了业务可读的信息。这一点在团队协作时尤其重要——不是每个收到告警的人都是写这个爬虫的人但格式化的信息能让任何接手的人快速理解发生了什么。5. 常见问题与排查实录5.1 机器人消息发不出去告警推送模块写完上线第一时间要做的不是等告警而是主动测试。我建议在代码里留一个--test参数跑一次真实推送。实测中我遇到过几类问题现在整理成速查表现象可能原因排查方式企业微信返回 errcode 93000Webhook 地址失效或机器人被删除到群里重新生成机器人替换配置企业微信返回 errcode 93001机器人被移出群聊重新添加机器人到群钉钉返回 errcode 300001加签参数错误或时间戳偏差过大检查签名算法和服务器的系统时间是否准确钉钉返回 errcode 310000消息内容含敏感词或格式不合法检查消息内容去掉特殊字符确认长度限制requests 报 SSL 错误服务器证书链问题或系统时间不对更新 ca-certificates同步时间推送超时目标服务器网络不通或代理配置异常先 curl 测试 webhook 是否可达这里单独提一下时间戳的问题。加签模式下钉钉会校验timestamp与服务器时间差是否在 1 小时内。如果服务器时间漂移严重签名判断就会失败。我遇到过一台 NTP 服务挂掉的服务器时间慢了两个小时所有钉钉告警全部发送失败故障恢复后那台机器的时间校准问题才消失。所以这类系统依赖时钟的一定要把 NTP 同步纳入日常巡检。5.2 告警刷屏与重复告警告警刷屏的根源往往是去重逻辑没做好。最容易出现的情况是爬虫任务失败后错误被异常捕获然后在一个大循环里重复上报。比如你的爬虫是分页抓取每一页失败都会抛一个异常如果你在每页的异常处理里都调用一次alert_client.send()那一次任务失败就能刷出几十条告警。解决思路有两个层面。第一层是在爬虫层面做错误聚合一整个任务不管失败多少次只发一次告警错误信息里带上失败页数和最早/最新的错误样例。第二层是在告警客户端层面做时间窗口去重也就是前面代码里self._recent_alerts那个字典。另外还有一个告警升级的逻辑可以加第一次失败发 WARN第 N 分钟还没恢复发 ERROR超过 1 小时发 CRITICAL。这种升级式告警比一上来就 CRITICAL 体验好得多因为很多问题确实能在几分钟内自动恢复没必要搞得全组紧张。5.3 爬虫长期运行的稳定性坑点监控系统做得再好也架不住爬虫本身埋的雷。这里说几个我在长期运维中反复遇到的坑。第一个坑是日志文件无限膨胀。爬虫如果用了print()调试输出重定向到日志文件之后文件会越来越大最后磁盘满爬虫写不了数据直接闪崩。解决办法是日志轮转用 Python 的logging.handlers.RotatingFileHandler设置单文件大小和备份数量比如单文件 50MB、保留 5 个备份。第二个坑是数据库连接泄漏。如果爬虫在循环里反复创建 SQLAlchemy session 而不关闭连接池最终会被耗尽导致数据库连接数超限的假性故障。这个问题的现象很迷惑人因为数据库本身没挂但新请求全部排队。解决办法是with sessionmaker() as session:这种上下文管理器用法确保 session 用后必关。第三个坑是内存缓慢增长。爬虫在长时间运行中如果某些大对象没有及时释放会逐渐吃光内存。这种问题监控里很难自动发现因为进程没死只是越来越慢。我的处理办法有两个方向一是用tracemalloc做定期快照对比内存分配热点二是更务实地给爬虫设置跑到一定时间自动重启的策略比如单进程最多运行 6 小时到点主动退出由调度平台重新拉起。对于很多爬虫场景定期重启是性价比极高的稳定性方案。第四个坑是反爬策略变化导致的假死。目标网站可能突然让你所有请求都返回 200 但内容是验证码页面此时爬虫不会报错反而会抓回一堆解析不了的数据。这种问题只有数据级监控能发现——对比历史抓取条数、字段非空率一旦指标异常下跌就立刻告警。这也是我前面反复强调数据级监控的原因它往往是最后一道、也是最关键的一道防线。6. 最后的几点体会这套系统从最初的只在任务失败时发一条企业微信消息演进到现在的四层指标 多级告警 降噪聚合 数据基线校验中间改了很多轮。我自己最大的体会是监控告警系统的价值不在于技术多复杂而在于敢不敢在半夜放心睡觉。当你把任务状态、数据异常、资源水位、推送通道全都打通并验证过之后那种踏实感是写多少行爬虫代码都换不来的。如果你想快速起步建议不要一口气追求完美。先做最小闭环一个爬虫任务、一次状态上报、一个企业微信机器人、一条失败告警。跑通之后再继续加数据量校验、加告警降噪、加调度平台。我个人的经验是80% 的收益来自 20% 的投入把最基本的失败告警和数据异常告警做好就已经超过了大多数裸奔的爬虫项目。最后再分享一个小技巧给告警机器人建一个专用的测试群和正式业务群分开。每次改动告警模板或推送逻辑先往测试群发一条确认格式和内容都没问题再切到正式环境。我因为图省事直接在正式群里调告警消息格式结果格式错误的消息把全组人都震了一遍之后学乖了测试群这个习惯一直保留到现在。