企业官网建设流程全解析

干爬虫这行最怕的不是被对方封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 aiohttp==3.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( connector=aiohttp.TCPConnector(limit=50, limit_per_host=10), timeout=aiohttp.ClientTimeout(total=15, connect=5), headers={"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) ..."} )

limit=50表示全局最多50个并发连接，limit_per_host=10表示同一域名最多10个并发。这两个参数是反反爬的“物理防线”，不只在代码层面限制，也在行为层面避免你表现得像个抓取机器人。

2.3 从异步请求到采集闭环的最小示例

我拿一个简化的舆情API来做演示，URL 结构是https://api.example.com/stream?since_id=12345。这个接口支持游标分页，返回 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，单请求约 200KB，1Mbps 带宽理论可承载 0.125MB/s。留给带宽 50% 余量，则每秒请求数约 0.3 个，并发控制在50左右比较合理。带宽、CPU、数据库连接池、目标站点的承载能力，这四个维度哪个先到瓶颈，就以哪个为准设信号量。

3. 实时采集核心模块的工程化实现

3.1 任务队列与URL去重策略

实时系统的任务队列我直接用asyncio.Queue，它天然适配生产者消费者模型。一个生产者往队列里塞URL，多个消费者从队列里取URL去下载，URL 的“待抓取集合”放在set里做内存去重。

seen_urls = set() q = asyncio.Queue(maxsize=10000) 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_retry：

async def fetch_with_retry(session, url, max_retries=3): 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:05+08:00"，想存进数据库最好统一成 UTC 的datetime对象，或者统一转成int时间戳。不然以后做聚合统计的时候，时区问题会让你想骂人。

3.4 增量更新：游标、时间戳、去重三次校验

实时爬虫和批爬虫最大的区别在于增量更新。批爬虫可以每次都全量拉，实时系统受限于接口压力和带宽，每次必须只拿“上次之后新增的数据”。

增量更新有三种常见策略：

• 基于时间游标：接口返回里带last_modified或publish_time，下次请求时把最大时间传回去。
• 基于ID游标：接口按自增ID排序，下次请求传?since_id=xxx。
• 基于分页深度的增量：这种方法不可靠，因为数据会滑动，容易漏抓和重抓。

我强烈建议优先用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/xhtml+xml,application/xml;q=0.9,image/webp,*/*;q=0.8 Accept-Language: zh-CN,zh;q=0.9,en;q=0.8 Accept-Encoding: gzip, deflate, br Connection: keep-alive

Accept-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 -= 1

IP池轮换是真实多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, mode="a") as f: await f.write(json.dumps(obj, ensure_ascii=False) + "\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 title=VALUES(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 高频问题速查表

写在最后

这套异步实时采集方案我在几个长期跑着的采集服务里用得很稳，核心就十二个字：事件循环驱动、限流控制节奏、增量持久化游标。异步解决并发吞吐，反反爬策略解决稳定生命周期，游标解决数据完整性。三条线互相咬合，缺一条都会在长跑中出问题。

如果你现在正准备上手写一个实时采集系统，我给你的建议是别急着跑demo，先花半天时间把架构图画清楚，明确数据从哪个接口来、并发控制在多少、失败了怎么重试、重启了游标怎么恢复。这四个问题想清楚，后面写代码都是水到渠成。踩过几次坑之后你就会发现，真正难的从来不是调通一个请求，而是让整个系统在任何异常情况下都不会丢数据、不会被打死。