干爬虫这行最怕的不是被对方封IP而是你辛辛苦苦把代码写完跑了一晚上定时任务第二天一看数据全是昨天的“过期新闻”。实时数据抓取和批量抓取是两个完全不同的游戏批量抓取拼吞吐量实时抓取拼延迟、稳定性和并发调度。你需要的不是每天cron跑一次的脚本而是一个事件循环驱动的异步采集通道外加一套能扛住封锁的反反爬策略组合拳。这篇文章我会从架构设计讲起一直到异步下载器、任务去重、增量更新、反反爬工程化落地、数据入库最后附上我实际踩过的坑和排查思路。用到的技术栈是 Python 3.10、asyncio、aiohttp、正则解析、异步数据库写入目标读者是已经写过简单爬虫、想往实时采集方向进阶的同学以及需要把爬虫做成稳定服务的一线开发。1. 实时爬取项目的整体设计思路1.1 实时抓取和定时抓取的本质区别大部分人第一次写爬虫都是这样一个while True或者 cron 表达式每隔一小时批量拉一次列表页再逐条进详情页解析。这种做法做成日报、周报没问题但一旦要跟进价格波动、舆情话题、库存变化就完全不够用了。实时抓取的本质不是“更快地跑循环”而是“在数据刚出现的那一刻就把它抓到”。这要求你的代码具备两个能力第一能维持长期运行的长连接而不是频繁“断开再握手”的短生命周期请求第二能同时保持成百上千个在途请求让网络IO的等待时间被其他任务填满。举个例子你抓一个行情接口平均响应时间 200ms。如果同步请求每秒最多5个请求。换成异步并发100个请求理论上每秒可以完成500个。这就是实时系统最基本的吞吐量杠杆。你不需要把单次请求变快你要的是让单线程在等待网络数据时不再空转。1.2 架构拆解生产者、消费者与事件循环实时爬虫的标准架构我会拆成四层任务生产层根据规则生成待抓取的URL列表支持定时触达、消息队列触发、接口推送触发。下载调度层基于 asyncio 的事件循环控制并发量、超时、重试、代理轮换。解析提取层从响应体里抽结构化字段做清洗、格式化、去重。存储分发层写文件、写数据库或者推送到下游消息队列。前三层之间用 asyncio.Queue 串起来第四层用连接池批量写入。这四层各司其职避免“下载和解析写成一坨、想加个去重就得重构”的尴尬局面。实时爬虫最容易翻车的地方不在下载而在任务生产速度和消费速度不匹配——下载太快解析跟不上内存里的队列越堆越大最后 OOM。所以每一层之间必须有明确的上下流控制和缓冲上限。1.3 技术选型为什么是 asyncio 加 aiohttp选 asyncio 不是因为别的库不好。requests是同步阻塞的你要并发就得开线程池每开一个线程都要付出创建和切换的代价scrapy自带 twisted 框架功能很强但部署、扩展门槛高做一个轻量级实时采集服务多少有点拎着牛刀上街。asyncio aiohttp足够轻又保留了完整的HTTP能力302 跟随、Cookie 存储、自定义 Headers、代理配置都支持而且你完全掌控事件循环想停想续都是一行代码的事。需要强调一点异步不是银弹。如果你的瓶颈在CPU密集的正则解析而不是网络等待一个事件循环解决不了问题。这种情况我会把解析任务用asyncio.to_thread扔进线程池网络IO走异步CPU解析走线程两边互不阻塞。2. 环境准备与核心依赖构建2.1 版本选择与安装我用 Python 3.10 以上版本原因很直白3.10 开始asyncio的 API 稳定TaskGroup、asyncio.timeout这类特性写起来舒服很多。3.8 也能跑但很多新特性得自己封装没必要给自己找麻烦。依赖我尽量控制在四个pip install aiohttp3.9.* pip install lxml pip install aiofiles pip install aiomysql如果你不想连数据库只写文件aiomysql可以换成标准库自带的asyncio队列加aiofiles。这里我不建议一上来就上 scrapy、selenium 全家桶先把最小闭环跑通后面哪儿痛再补哪儿。2.2 构建一个可复用的异步会话对象aiohttp 里最容易被忽视的就是ClientSession的复用。很多人写异步爬虫会犯一个错每请求一次就创建一个session用完了再关。这等于每次都要重建 TCP 连接、TLS 握手异步省下的时间全浪费在握手上了。正确做法是启动时创建一个ClientSession整个进程生命周期复用并配置好连接池import aiohttp session aiohttp.ClientSession( connectoraiohttp.TCPConnector(limit50, limit_per_host10), timeoutaiohttp.ClientTimeout(total15, connect5), headers{User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) ...} )limit50表示全局最多50个并发连接limit_per_host10表示同一域名最多10个并发。这两个参数是反反爬的“物理防线”不只在代码层面限制也在行为层面避免你表现得像个抓取机器人。2.3 从异步请求到采集闭环的最小示例我拿一个简化的舆情API来做演示URL 结构是https://api.example.com/stream?since_id12345。这个接口支持游标分页返回 JSON 数组内部包含id、title、content、publish_time四个字段。import asyncio import aiohttp import json async def fetch(session, url): async with session.get(url) as resp: if resp.status 200: return await resp.json() return []这是最基础的雏形。但真实工程里你要考虑的问题远不止“能不能拿到数据”拿不到的时候要不要重试重试会不会把对方服务器打挂拿到之后解析失败怎么办每个问题都值得单独拆开讲。2.4 并发控制信号量不是可有可无的选项asyncio.Semaphore是并发控制的闸门。你可以在session的连接池设置全局并发上限但那是TCP层限制业务层的“同一时刻最多跑多少个任务”得用信号量。async def worker(sem, session, url): async with sem: return await fetch(session, url)信号量设置多少比较合适没有标准答案。我会根据“对方接口的响应时间”和“我方单机带宽”来估算。比如对方响应平均 200ms单请求约 200KB1Mbps 带宽理论可承载 0.125MB/s。留给带宽 50% 余量则每秒请求数约 0.3 个并发控制在50左右比较合理。带宽、CPU、数据库连接池、目标站点的承载能力这四个维度哪个先到瓶颈就以哪个为准设信号量。3. 实时采集核心模块的工程化实现3.1 任务队列与URL去重策略实时系统的任务队列我直接用asyncio.Queue它天然适配生产者消费者模型。一个生产者往队列里塞URL多个消费者从队列里取URL去下载URL 的“待抓取集合”放在set里做内存去重。seen_urls set() q asyncio.Queue(maxsize10000) async def produce(url): if url not in seen_urls: seen_urls.add(url) await q.put(url) async def consume(): while True: url await q.get() try: data await fetch_one(url) finally: q.task_done()seen_urls只增不减长时间跑会有内存压力。我的经验是带着时间戳去重seen_urls里存(url, timestamp)每隔一段时间清理超过30分钟的老记录这样既保证短时间内的去重又不会让内存无限增长。如果你的项目需要跨重启去重就换成 Redis 的SETNX天然带过期时间。3.2 下载器封装响应码判断、超时与重试机制下载器是一个高频使用、必须反复打磨的模块。我通常会封装一个带重试的fetch_with_retryasync def fetch_with_retry(session, url, max_retries3): for attempt in range(max_retries): try: async with session.get(url) as resp: if resp.status 200: return await resp.text() if resp.status in (403, 429): await asyncio.sleep(2 ** attempt) continue except (aiohttp.ClientError, asyncio.TimeoutError) as e: if attempt max_retries - 1: raise await asyncio.sleep(0.5 * (attempt 1)) return None要特别说明重试语义429和403是目标主动限流重试间隔要拉长采用指数退避而不是固定间隔TimeoutError是网络抖动可以快速重试。重试逻辑如果写反了比如遇到 403 立刻重试10次只会加速封禁。另外重试次数不建议超过3次超过之后就直接把它丢进“失败Topic”留着人工排查不要在实时链路上无限阻塞。3.3 数据解析与字段抽取的细节处理解析模块我平时用lxml的etree.HTML和正则配合使用HTML 用 XPath 定位结构JSON 直接用json.loads。这里有一个容易翻车的点很多接口返回的 JSON 并不是纯 JSON而是var data {...}或者列表外面套了注释。你需要在解析前做一层“清洗”用正则把有效负载切出来。import re import json def extract_json(payload: str) - dict: match re.search(r\{.*\}, payload, re.S) if match: return json.loads(match.group()) raise ValueError(No JSON object found)字段抽取时要顺手格式化时间戳。比如接口返回的是2024-08-10T15:04:0508:00想存进数据库最好统一成 UTC 的datetime对象或者统一转成int时间戳。不然以后做聚合统计的时候时区问题会让你想骂人。3.4 增量更新游标、时间戳、去重三次校验实时爬虫和批爬虫最大的区别在于增量更新。批爬虫可以每次都全量拉实时系统受限于接口压力和带宽每次必须只拿“上次之后新增的数据”。增量更新有三种常见策略基于时间游标接口返回里带last_modified或publish_time下次请求时把最大时间传回去。基于ID游标接口按自增ID排序下次请求传?since_idxxx。基于分页深度的增量这种方法不可靠因为数据会滑动容易漏抓和重抓。我强烈建议优先用ID游标或时间游标。每次处理完一批把最大游标存下来放到 Redis 或本地文件进程重启后接着跑不丢数据。另外同一份数据可能被推送到多个渠道数据库里需要建唯一索引。比如UNIQUE KEY uk_source_id (source, item_id)入库时用INSERT ... ON DUPLICATE KEY UPDATE这样就算幂等失败了也不会产生脏数据。4. 反反爬策略的工程化落地4.1 目标服务器是怎么识别“你”的反爬不是一个具体的功能而是一套“行为画像”匹配机制。服务器那边通常从四个维度判断你是不是真人请求头User-Agent、Accept-Language、Referer是否符合真实浏览器语义。行为模式请求频率是否稳定得像时钟还是带有人类随机的抖动。Cookie 状态新会话是否先访问首页、再带 Cookie 访问数据接口。设备指纹TLS 指纹、HTTP/2 指纹、浏览器画布指纹。你做得越像真人的浏览路径被封的概率就越低。这不是让你去对抗什么而是让你的采集行为更符合公开网络的文明规则。4.2 Headers 伪装别只改User-Agent初学者最容易犯的错就是只改User-Agent。一个完整的浏览器请求头通常长这样User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 Accept: text/html,application/xhtmlxml,application/xml;q0.9,image/webp,*/*;q0.8 Accept-Language: zh-CN,zh;q0.9,en;q0.8 Accept-Encoding: gzip, deflate, br Connection: keep-aliveAccept-Encoding如果写成identity服务器直接知道你是脚本。还有Referer访问详情页时如果不带列表页的 Referer很多站点的安全策略会直接拒绝。我的建议是把一组常用浏览器的UA放进列表每次请求随机取一个同时把Accept、Accept-Language固定成浏览器的默认值别偷懒。4.3 Cookie 与会话保持的管理细节如果你的采集目标是登录后才能看的数据先说一句合规红线先确认好。只讲技术实现的话aiohttp.ClientSession自带cookiejar发起登录请求拿到 cookie 之后不要每次手动塞 Header而是让 session 自动维护。async def login_and_build_session(login_url, username, password): session aiohttp.ClientSession() await session.post(login_url, json{username: username, password: password}) return session登录后的 cookie 通常会过期你需要定期心跳或者捕获401/403后重新登录。比较稳妥的做法是封装一个ReauthSession类请求返回 401 时自动重登一次再重新请求。注意重登逻辑要加锁否则并发请求同时触发登录容易把账号顶掉。4.4 频率控制与IP池轮换的正确姿势频率控制的工程标准是“单位时间窗口内的离散均匀”。我见过有人用asyncio.sleep(1)当节流看似每秒一个请求其实因为事件循环的时间偏差实际分布会形成微小的簇。想要平滑控制最好用令牌桶算法class RateLimiter: def __init__(self, rate): self.rate rate self.tokens rate self.updated asyncio.get_event_loop().time() async def acquire(self): now asyncio.get_event_loop().time() self.tokens (now - self.updated) * self.rate self.updated now if self.tokens self.rate: self.tokens self.rate if self.tokens 1: delay (1 - self.tokens) / self.rate await asyncio.sleep(delay) self.tokens 0 else: self.tokens - 1IP池轮换是真实多IP采集时需要管理的资源注意这里说的是合规的代理服务。轮换策略最笨也最有效的是“按请求轮换、按失败自动摘除”。每次请求前从池里取一个可用IP如果失败连续3次就从池里摘掉等冷却期后再放回。这里要敲一个警钟不管代理池多优质都不要为了“证明并发能力”把并发调到几百上千。目标站的带宽和数据库又不是你家开的单机无脑高并发只会让整个IP段被屏蔽到时候换再多代理也救不回来。4.5 碰到验证码怎么办先停不是先绕我收到过不少私信问“怎么过验证码”。我的态度一直很明确公开数据采集遇到验证码说明对方从业务层面明确不欢迎程序化访问这个时候应该停下来重新评估采集方案而不是研究怎么绕。如果你确实需要持续性采集优先看对方有没有官方API、开放数据接口、RSS或数据授权渠道。这些方案一劳永逸没有任何风控风险。如果非结构化页面本身没有API且数据价值足够高也可以考虑合法授权后再采集。不要为了短期数据把整条业务线的IP信誉赔进去。5. 数据落地与下游实时消费5.1 写入数据库还是落盘文件数据量不大每秒几十条的时候直接aiofiles.open()逐行写 JSONL 是成本最低的方案import aiofiles async def write_line(path, obj): async with aiofiles.open(path, modea) as f: await f.write(json.dumps(obj, ensure_asciiFalse) \n)数据量上了每秒几千条文件写入就会变成瓶颈这时候得换数据库或消息队列。写入数据库推荐异步批量入库每条数据都await conn.execute()的做法效率极低。正确姿势是攒一批再写batch [] async def batch_insert(conn, item): batch.append(item) if len(batch) 200: await flush(conn) async def flush(conn): async with conn.cursor() as cur: await cur.executemany( INSERT INTO items (source, item_id, title, content, publish_time) VALUES (%s, %s, %s, %s, %s) ON DUPLICATE KEY UPDATE titleVALUES(title), batch ) batch.clear()executemany批量插入比逐条快接近一个数量级还能顺手解决幂等去重。5.2 背压控制与“积水”问题生产速度大于消费速度时队列会膨胀内存暴涨。这就是背压问题。asyncio.Queue(maxsize)只是挡住“入队”操作并不会自动放慢生产者。你需要在生产者侧做感知if q.qsize() 8000: await asyncio.sleep(0.5)或者更优雅的做法await q.put()在队列满时会阻塞这天然就把生产者限住了。我倾向于让生产者主动感知队列水位动态调整抓取任务的暂停时间。这样链路不会断裂数据也不会丢。5.3 与下游分析系统的对接方式实时采集很多时候不是终点而是起点。采集后的数据要送进统计系统、告警系统、推荐引擎。最简单的对接方式是输出到消息队列比如用aioamqp或直接用 Redis 的 List/Stream下游系统消费解读。如果你暂时不想引入消息队列用数据库表加一个processed_flag字段也是可以的。采集进程负责写分析进程负责更新状态。缺点是数据库查询变重不建议在数据量大时这么干。我的偏好是采集层只负责“干净地拿到数据并通过队列移交”绝不负责复杂的业务聚合保持管线单一职责。6. 实际踩坑与排查记录6.1 asyncio 事件循环被阻塞任务全部卡住有段时间我发现采集任务跑着跑着就全部停住事件循环像是“死”了。排查后发现是解析模块里用了一个同步的time.sleep(2)这一睡直接阻塞了事件循环所有异步任务跟着一起休眠。事件的教训是异步代码里不能出现任何阻塞调用。time.sleep换成asyncio.sleep同步正则如果数据量极大就包一层await asyncio.to_thread。后来我把整个下载器里的阻塞函数排查了一遍凡是同步IO全部转移线程池事件循环再无卡顿。6.2 并发拉满后连接被重置误伤整段IP有一次我把并发调到200结果三分钟后所有请求全部报ConnectionResetError然后整个出口IP段被目标服务断连。这属于典型的“物理层过载惩罚”。后面我总结出一条铁律新域名上线并发从10开始每5分钟递增10直到出现第一条非200响应就回退20%。这种做法在目标没有明确 API 限流的情况下能最大程度避免触雷。6.3 响应内容乱码编码声明只是参考JSON 接口返回的charset经常是错的有的声明utf-8实际是gbk有的声明gb2312实际是gb18030。我踩过一次解析全变成了乱码排查了半天发现是编码错配。解决办法是用chardet先探测再解码。aiohttp的resp.text()默认使用响应头里的编码遇到不确定的应该用await resp.read()拿原始字节再用chardet.detect()判编码最后decode()。这个步骤只加一次但能救你无数次。6.4 内存持续增长对象引用堆积长时间运行后内存曲线一直向上走最典型的原因是“引用了不该引用的对象”。比如把每次请求的session都存进 list或者queue.task_done忘记调用队列任务永远无法被标记完成积压在内部。还有个容易忽略的坑日志里拼接字符串太多logging底层把旧日志引用挂在内存。长期运行的程序日志要用RotatingFileHandler滚动写别让单文件无限增长。6.5 高频问题速查表现象可能原因排查步骤解决方案任务全部卡住不执行同步阻塞操作堵住事件循环检查有没有 time.sleep、同步文件读换成 asyncio 对应异步方法请求大量超时并发过高导致连接池耗尽看 aiohttp 错误一栏是否全是 timeout降低信号量或换代理池返回 403Headers 不完整或 Cookie 失效打印响应头和页面标题确认补全 Accept/Referer重新登录拿到数据全是乱码字符编码声明错误用 chardet 检测用resp.read()后手动 decode内存持续增加Queue 任务未完成或日志堆积看队列深度和 GC 对象计数task_done补齐、日志滚动数据漏抓游标未更新或分页滑动对比上次游标和数据库最大值改用 ID/时间游标并持久化当前位置写在最后这套异步实时采集方案我在几个长期跑着的采集服务里用得很稳核心就十二个字事件循环驱动、限流控制节奏、增量持久化游标。异步解决并发吞吐反反爬策略解决稳定生命周期游标解决数据完整性。三条线互相咬合缺一条都会在长跑中出问题。如果你现在正准备上手写一个实时采集系统我给你的建议是别急着跑demo先花半天时间把架构图画清楚明确数据从哪个接口来、并发控制在多少、失败了怎么重试、重启了游标怎么恢复。这四个问题想清楚后面写代码都是水到渠成。踩过几次坑之后你就会发现真正难的从来不是调通一个请求而是让整个系统在任何异常情况下都不会丢数据、不会被打死。