从能跑到能稳定跑很久——爬虫的工程化跃迁
前几讲,已经能写出一个会发请求、会解析、会绕反爬的爬虫了。但它还只是个"脚本"——点一下运行,它跑完就结束。
而真实项目里的爬虫要做的事完全不一样:
- 24 小时不停跑,有时候要连续跑几个月
- 要抓几百家公司、几十万个职位
- 中途网络断了、电脑重启了、被封了,得能自己恢复
- 抓到的数据要进数据库、要去重、要追踪状态
- 已经入库的数据过期了要清洗、链接失效了要回写
这些事情不是写代码能力的问题,而是工程设计的问题。
这一讲需要来回答最后一个核心问题:怎么把一个会跑的脚本,变成一个能稳定跑很久的系统?
学完这一讲,回头看猎聘项目里那个 1000 行的 spider_liepin_company.py,你会发现它其实没什么神秘的——它只是把这一讲讲的几件事,**每一件都做到了"豪华版"**而已。
一、从脚本到系统:核心矛盾在哪里?
把脚本变成系统,会遇到 5 个绕不开的问题。
一个一个解决。
问题1:配置硬编码 → 换个站点要改一堆代码 → 配置驱动
问题2:跑到一半崩了 → 重新跑一遍前面白做 → 状态系统
问题3:抓的数据越来越多 → 单进程跑不完 → 任务切片 + 多实例
问题4:数据怎么用?怎么去重? → 不能堆在硬盘 → 入库链路
问题5:已入库数据会过期、失效 → 数据库会脏 → 清洗回写
这 5 个问题对应着工程化爬虫的 5 个核心模块。逐个拆开讲。
二、问题 1:配置驱动 —— 把"会变的"和"不变的"分开
我们在第 4 讲写的爬虫,所有 URL、字段名、目录路径都直接写死在代码里:
URL = "https://realpython.github.io/fake-jobs/"
DATA_DIR = "data/jobs"
这个写法的问题:换一个站点要抓,整个代码都要改。换 10 个站点,意味着要写 10 份几乎一样的代码。
正确做法:把"会变的"抽出来放到配置文件里。
2.1 用 INI 配置文件
# config/spider_liepin.ini
[spider]
chromePath = /usr/bin/chromium
savePath = ./data/liepin
maxSpiderCount = 500
timeInterval = 86400
[api]
getProxy = http://proxy-pool.internal/get
jobUrl = http://api.internal/jobparse
comUrl = http://api.internal/comparse
代码里:
import configparser
cfg = configparser.ConfigParser()
cfg.read("config/spider_liepin.ini", encoding="utf-8")
chrome_path = cfg["spider"]["chromePath"]
save_path = cfg["spider"]["savePath"]
max_count = int(cfg["spider"]["maxSpiderCount"])
get_proxy_api = cfg["api"]["getProxy"]
这就是 spider.py 第一步做的事:spider.ini 装载所有"环境相关"的东西——浏览器路径、代理接口、入库接口、最大抓取数、去重时间窗口。
2.2 命令行参数
但有些东西,连配置文件都嫌死板——比如"今天我只想跑第 50 家公司"、"我要切到生产环境跑一下"。这种临时性指令用命令行参数:
import argparse
parser = argparse.ArgumentParser()
parser.add_argument("-m", "--mode", default="ann", help="运行模式: ann/cjob/cp/...")
parser.add_argument("-f", "--shard", type=int, default=0, help="分片编号")
parser.add_argument("-d", "--env", default="dev", help="环境: dev/prod")
parser.add_argument("-c", "--company", help="单公司定向跑")
args = parser.parse_args()
if args.mode == "ann":
run_ann_spider(shard=args.shard, env=args.env)
elif args.mode == "cjob":
run_cjob_spider(shard=args.shard, env=args.env)
启动:
python main.py -m cjob -f 50 -d prod
python main.py -m cjob -c "腾讯" -d dev
这就是通用结构化系统里 main.py 那套 -m / -f / -d / -c 参数的全貌。一个脚本通过参数变成多种运行模式,省下了写 10 个脚本的功夫。
2.3 配置层级:default + override
更进阶的做法是配置叠加:
setting_default.ini ← 全局默认值
↓ 被覆盖
setting_sch_99.ini ← 第 99 个分片的特定配置
↓ 被覆盖
命令行参数 -d prod ← 临时覆盖
读取顺序:先读默认,再读特定分片,最后命令行覆盖。这样既能保持默认值统一,又能让每个分片有自己的个性化设置。通用结构化系统采用的就是这套层级。
三、问题 2:状态系统 —— 让爬虫"记得自己抓到哪了"
第 4 讲我们用 progress.txt 记录"抓到第几条"。但真实项目里,事情远比这复杂:
- 一个公司有上千个职位,要按"地区 × 经验 × 薪资"切成上百个组合,每个组合都得记录是否抓完
- 程序中途崩了,状态文件不能损坏(否则全完了)
- 老格式的状态文件要平滑升级到新格式,不能丢数据
这就是猎聘公司版那套 spider_status_<ComId>.json 解决的问题。
3.1 状态文件的本质:就是个 JSON
{
"ComId": "12345",
"ComName": "腾讯",
"crawled_conditions": [
"010|java|3-5|15-25",
"010|java|5-10|15-25",
"010|python|3-5|15-25"
],
"total_conditions": 200,
"last_update": "2026-05-06T14:30:00"
}
读起来是一个数组,记录了"哪些条件组合已经抓完了"。
3.2 关键操作:判断 / 标记 / 完成率
import json, os
def load_status(com_id):
path = f"status/spider_status_{com_id}.json"
if not os.path.exists(path):
return {"crawled_conditions": []}
with open(path, encoding="utf-8") as f:
return json.load(f)
def is_crawled(status, condition_key):
return condition_key in status["crawled_conditions"]
def mark_crawled(status, condition_key):
if condition_key not in status["crawled_conditions"]:
status["crawled_conditions"].append(condition_key)
def is_company_done(status, total):
return len(status["crawled_conditions"]) >= total
主流程:
status = load_status(com_id)
all_conditions = generate_conditions(com_id) # 200 个组合
for cond in all_conditions:
cond_key = f"{cond['city']}|{cond['type']}|{cond['exp']}|{cond['salary']}"
if is_crawled(status, cond_key):
continue # 这个组合已经抓过了,跳过
crawl_one_condition(cond)
mark_crawled(status, cond_key)
save_status(com_id, status) # 每抓完一个组合就保存
if is_company_done(status, len(all_conditions)):
print(f"公司 {com_id} 已全部抓完,进入增量模式")
这就是 __is_condition_crawled() / __mark_condition_crawled() 的核心逻辑。
3.3 为什么要"原子写入"?
普通的 open(path, "w") + write() 有个隐患:写到一半程序崩了,文件就坏了。下次启动读不出来,所有状态丢失。
工程化做法:先写临时文件,再原子重命名。
import json, os, tempfile
def save_status_safe(path, data):
# 1. 备份现有文件
if os.path.exists(path):
os.replace(path, path + ".backup")
# 2. 写到临时文件
tmp = path + ".tmp"
with open(tmp, "w", encoding="utf-8") as f:
json.dump(data, f, ensure_ascii=False, indent=2)
f.flush()
os.fsync(f.fileno()) # 强制写到磁盘,不在系统缓冲区里
# 3. 验证临时文件能正常读
with open(tmp, encoding="utf-8") as f:
json.load(f) # 能解析就说明完整
# 4. 原子重命名(操作系统保证这一步要么成功要么不发生)
os.replace(tmp, path)
这套流程是从"血的教训"里来的——猎聘公司版项目代码里专门写明了:遇到过状态文件损坏问题,才把写入流程加固到这个程度。
3.4 格式兼容与迁移
项目跑久了,会想升级状态文件格式(比如从字典换成数组以提高性能)。但线上几千个老文件不能丢。
def load_and_migrate(path):
with open(path, encoding="utf-8") as f:
raw = json.load(f)
# 旧格式: {"crawled": {"key1": True, "key2": True}}
if isinstance(raw.get("crawled"), dict):
new_data = {
"crawled_conditions": list(raw["crawled"].keys()),
"version": "v2",
}
save_status_safe(path, new_data)
return new_data
# 新格式直接返回
return raw
这就是 convert_existing_status_files() 干的事——读老文件 → 自动转新格式 → 落盘。线上系统最值钱的不是代码,是历史数据,这种平滑迁移能力是工程化系统的标志。
四、问题 3:任务切片 —— 一个爬虫干不动,就开多个
4.1 为什么要切片?
一家大公司可能有 5000 个职位。一次性请求列表接口,只能拿到前 100 条(深翻页都受限制)。怎么办?
把"大查询"切成"小查询"——按地区、经验、薪资、职类做笛卡尔积:
cities = ["北京", "上海", "深圳", "杭州"]
exps = ["1-3年", "3-5年", "5-10年"]
salaries = ["10-20k", "20-40k", "40k+"]
conditions = [
{"city": c, "exp": e, "salary": s}
for c in cities for e in exps for s in salaries
]
# 4 × 3 × 3 = 36 个条件组合
每个组合大约返回几十到几百条职位,加起来覆盖全公司,每个组合也都不会触发深翻页限制。这就是猎聘公司版"四维条件组合"的本质——[dq, jobtitle, workyear, salary]。
4.2 阈值优化:小结果集不切
如果一个公司只有 50 个职位,你切 36 个组合就是浪费——每个组合都要发请求、都要落盘。做个阈值判断:
total = get_total_count(company)
if total <= 120:
# 直接抓一轮
crawl_simple(company)
else:
# 切片
for cond in generate_conditions(company):
crawl_one(cond)
这就是猎聘代码里那个 "120 阈值" 的由来——基于实际数据观察出的拐点。
4.3 多实例分片:把任务分给多台机器
更大的体量,要靠多机并行:
# 机器 1
python main.py -f 1 # 跑分片 1(公司 ID 模 10 == 1)
# 机器 2
python main.py -f 2
# 机器 10
python main.py -f 10
代码里:
def get_companies_for_shard(shard, total_shards=10):
all_companies = fetch_all_companies()
return [c for c in all_companies if int(c["id"]) % total_shards == shard]
就这么简单。pyexe.py 那个生成 .bat 文件的脚本,做的就是把 python main.py -f 1、-f 2、-f 3...批量生成多个启动批处理,部署到多台机器上。
4.4 时间调度:不要 24 小时不停打
import datetime, time
def wait_until(hour):
now = datetime.datetime.now()
target = now.replace(hour=hour, minute=0, second=0)
if target <= now:
target += datetime.timedelta(days=1)
seconds = (target - now).total_seconds()
print(f"休眠 {seconds/3600:.1f} 小时,等到 {hour} 点")
time.sleep(seconds)
while True:
if datetime.datetime.now().hour < 8:
wait_until(8) # 凌晨不抓
run_spider()
wait_until(9) # 跑完等到下次 9 点
run_spider()
wait_until(17) # 再等到下次 17 点
这就是 spider_liepin_conditions.py 那套"凌晨 8 点前不抓、9 点和 17 点各跑一次"的实现。
为什么不 24 小时跑?
- 凌晨流量小,你抓得猛容易被识别
- 抓久了代理消耗大,适当休息能让代理池补货
- 业务有节奏——招聘网站本身上班时间发布的新职位多,夜里没多少新数据,抓也是浪费
五、问题 4:入库链路 —— "先文件、后接口"
抓到数据之后,不要直接往数据库写。这是工程化爬虫最重要的一条原则。
5.1 为什么不直接入库?
直接入库:
爬虫 → 数据库
↑
这一步出错,数据就没了
先文件后接口:
爬虫 → 本地 JSON → API → 数据库
↑ ↑
数据保住了 API 失败可以重试
好处:
- 可追溯:出问题了,能看到每条数据当时长什么样
- 可重跑:解析逻辑改了,直接重跑本地 JSON,不用重新抓网络
- 解耦:抓取和入库是两件事,各自可以独立扩展、独立调试
- 批量重试:某个 API 挂了 1 小时,这 1 小时的文件等 API 恢复了批量重传
5.2 完整链路:落盘 → 调接口 → 归档
import os, json, requests, shutil
def save_raw(job_data, save_path):
"""步骤1: 抓到的数据落盘"""
fname = f"{job_data['id']}.json"
path = os.path.join(save_path, "job", fname)
with open(path, "w", encoding="utf-8") as f:
json.dump(job_data, f, ensure_ascii=False)
return path
def call_parse_api(file_path, api_url):
"""步骤2: 调用解析入库接口"""
with open(file_path, encoding="utf-8") as f:
content = f.read()
resp = requests.post(api_url, data={
"comFrom": "liepin",
"fileName": os.path.basename(file_path),
"content": content,
"retry": 0,
}, timeout=30)
return resp.json()
def archive(file_path, success):
"""步骤3: 根据结果归档"""
today = datetime.date.today().isoformat()
target_dir = f"success/{today}" if success else f"failed/{today}"
os.makedirs(target_dir, exist_ok=True)
shutil.move(file_path, target_dir)
# 主流程
def enter_database(job_data):
file_path = save_raw(job_data, save_path)
try:
result = call_parse_api(file_path, "http://api.internal/jobparse")
if result.get("code") == 200:
archive(file_path, success=True)
return True
else:
archive(file_path, success=False)
return False
except Exception as e:
print(f"入库失败: {e}")
# 不归档,留在 job/ 目录,下次启动会重试
return False
5.3 目录就是状态机
data/
├── job/ ← 待入库
│ ├── 12345.json
│ └── 12346.json
├── success/ ← 已成功
│ └── 2026-05-06/
│ ├── 11111.json
│ └── 11112.json
├── failed/ ← 失败,要排查
│ └── 2026-05-06/
│ └── 99999.json
├── expired/ ← 业务拒收(过期/重复)
└── progress/ ← 状态文件
目录结构本身就是任务的当前状态。看一眼 job/ 还有多少文件,就知道还有多少没入库;failed/ 有多少,就知道有多少要排查。这比写监控代码简单多了。
5.4 业务状态码的处理
API 不只返回"成功/失败",还可能返回业务码:
result = call_parse_api(...)
code = result.get("code")
if code == 200:
archive(file_path, success=True)
elif code in (301, 302, 303, 304, 305):
# 业务拒收:过期、重复、人工修改过等
move_to_expired(file_path) # 不算失败,不重试
elif code == 500:
# 真正的失败,留在 job/ 等下次重试
pass
else:
archive(file_path, success=False)
这就是通用结构化系统".ok / .err / .expired"三态文件机制的本质。
六、问题 5:清洗与回写 —— 数据库会"变质",得有人扫垃圾
6.1 为什么要清洗?
抓回来的职位入库之后,会过期:
- 招聘网站把这条职位下线了
- 招聘单位把岗位招满了
- 链接 404 了
数据库里如果还显示这些"僵尸职位",用户点进去什么都看不到——用户体验崩了。
所以要有一个专门的清洗爬虫,定期去验证已入库职位是否还有效。这就是 spider_yupao_clean.py 干的事。
6.2 清洗爬虫的特殊性
| 普通采集爬虫 | 清洗爬虫 |
|---|---|
| 目标:抓新数据 | 目标:验证老数据 |
| 字段:抓全 | 字段:不抓,只看页面状态 |
| 数据流:站点 → 数据库 | 数据流:数据库 → 站点 → 数据库(回写状态) |
| 规模:百万级 | 规模:千万级(存量更大) |
清洗爬虫的输入是数据库已有的链接列表,输出是过期标记。
def clean_job_links():
while True:
# 1. 从内部接口拉一批待验证的链接
links = fetch_links_to_verify(batch_size=100)
if not links:
break
for job_id, url in links:
status = check_job_status(url)
if status == "expired":
# 2. 调用过期回写接口
report_expired(job_id)
time.sleep(random.uniform(0.5, 1.5))
# 3. 写游标,下次从这里继续
save_cursor(links[-1][0])
def check_job_status(url):
page.goto(url)
# 各种判断条件
if "404" in page.title():
return "expired"
if page.query_selector(".job-ended"):
return "expired"
if page.query_selector(".verify-page"):
return "blocked"
return "valid"
def report_expired(job_id):
requests.post("http://api.internal/job_to_expired", json={"id": job_id})
6.3 清洗爬虫的工程化升级
清洗任务规模大、频率高,所以对性能极度敏感。spider_yupao_clean.py 加了一堆优化:
- 浏览器上下文复用:不为每个链接开新 context,而是复用同一个,节省启动开销
- LRU 缓存:同一批次里如果有重复 URL,直接用缓存结果
- 指数退避重试:网络抖动时重试,失败间隔逐步拉长
- 多选择器兼容:验证页可能有多种样式,用一组选择器轮询匹配
- 统一资源清理:用
with和上下文管理器保证浏览器实例最终一定被关掉
class JobLinkCleaner:
def __init__(self):
self.cache = {} # LRU cache
self.browser = None
self.context = None
def __enter__(self):
self.playwright = sync_playwright().start()
self.browser = self.playwright.chromium.launch(headless=True)
self.context = self.browser.new_context()
return self
def __exit__(self, *args):
self.context.close()
self.browser.close()
self.playwright.stop()
def check_with_retry(self, url, max_retry=3):
if url in self.cache:
return self.cache[url]
for i in range(max_retry):
try:
result = self._check_once(url)
self.cache[url] = result
return result
except Exception:
time.sleep(2 ** i)
return "unknown"
# 使用
with JobLinkCleaner() as cleaner:
for url in urls:
cleaner.check_with_retry(url)
这就是项目里"工程化程度最高的脚本"长什么样——它不是逻辑最复杂,而是对性能、稳定性、资源管理的考虑最周密。
七、问题之外:基类抽象 —— 把"共性"集中起来
到这里你应该发现,无论是哪个站点的爬虫,都要做同样的事:读配置、获取代理、判断去重、调入库接口、归档文件、记录日志。
如果每个站点脚本都重写一遍,维护成本爆炸。所以要抽出一个公共基类:
class Spider:
"""所有渠道爬虫的基类"""
def __init__(self, config_path):
self._load_config(config_path)
self._init_directories()
# ===== 通用能力 =====
def get_proxy(self):
"""从代理池领代理"""
...
def set_proxy_record(self, success_count, limited_type):
"""上报代理使用情况"""
...
def is_spider_today(self, item_id):
"""今天是否抓过"""
...
def job_exist(self, com_name, job_title, city):
"""数据库是否已存在"""
...
def com_exist(self, com_name):
"""公司是否已存在 / 是否在黑名单"""
...
def enter_database(self, file_path, parse_url):
"""通用入库链路:调接口 + 归档"""
...
def random_sleep(self, min_s=1, max_s=3):
"""随机休眠"""
...
# ===== 子类必须实现 =====
def crawl(self):
"""每个站点自己实现具体抓取逻辑"""
raise NotImplementedError
class LiepinSpider(Spider):
"""猎聘爬虫,只关注'怎么从猎聘抓到数据'"""
def crawl(self):
proxy = self.get_proxy() # 用基类
# ... 猎聘特定的抓取流程
for job in jobs:
if self.is_spider_today(job["id"]): # 用基类
continue
file_path = self.save_raw(job)
self.enter_database(file_path, self.job_url) # 用基类
class YupaoSpider(Spider):
"""鱼泡爬虫"""
def crawl(self):
# 鱼泡是移动端,实现不一样
# 但去重、入库、代理等共性都用基类
...
spider.py 那个公共基类,本质就是这件事:把"所有渠道都要做的事"集中,让子类只关注"这个渠道独有的事"。
最大的好处是改一处生效十处——比如代理池接口变了,只改基类的 get_proxy(),所有子类自动生效。这就是为什么真实项目里基类那么重要。
八、把所有模块串起来:工程化爬虫的全貌
讲到这里,我们可以把一个工程化爬虫的完整架构画出来了:
┌──────────────────────────────┐
│ 配置层 (.ini + CLI) │
│ 路径、接口、参数、模式 │
└────────────┬─────────────────┘
│
┌────────────▼─────────────────┐
│ 调度层 (main.py) │
│ 按模式分发、按分片切分 │
└────────────┬─────────────────┘
│
┌────────────▼─────────────────┐
│ 基类层 (spider.py) │
│ 代理 / 去重 / 入库 / 归档 │
└────┬──────────┬──────────┬───┘
│ │ │
┌────────▼──┐ ┌─────▼────┐ ┌───▼────┐
│ LiepinSp. │ │ YupaoSp. │ │ 51JobSp│
│ (子类) │ │ (子类) │ │ (子类) │
└────────┬──┘ └─────┬────┘ └───┬────┘
│ │ │
└─────┬────┴──────────┘
▼
┌──────────────────────────────┐
│ 落盘层 (data/) │
│ job/ → success/ / failed/ │
└────────────┬─────────────────┘
│
┌────────────▼─────────────────┐
│ 入库层 (内部 API) │
│ jobparse / comparse │
└────────────┬─────────────────┘
│
┌────────────▼─────────────────┐
│ 清洗层 │
│ yupao_clean / job_to_expired │
└──────────────────────────────┘
这就是猎聘 + 通用结构化系统的完整骨架。每一层都解决了一个特定问题,合起来就是"能稳定跑很久"。
九、本讲核心认知
"工程化"不是把代码写得复杂,而是把'每一种会出错的情况'都提前安排好
| 会出什么错 | 怎么预防 |
|---|---|
| 配置改了要重新发版 | 配置文件 + 命令行参数 |
| 跑一半崩了 | 进度文件 + 状态文件 + 原子写入 |
| 单机跑不动 | 分片 + 多实例 |
| 数据丢了 | 先文件后接口 + 三态归档 |
| 数据库变脏 | 清洗爬虫 + 回写过期 |
| 改一处要改十处 | 基类抽象 |
| 站点改版了 | 配置驱动 + 模板化抽取函数 |
回头去看 spider.py 的初始化代码,你会发现它不过是在做"读配置、建目录、初始化代理客户端、加载状态文件"——没有任何花哨的东西,只是把所有该做的事都做了。
组内导航：⬅️ 练手网站 | 🏠 00-爬虫课程六讲 | ➡️ 对照项目源码拆解——从配置到入库的完整链路
💬 评论