---
title: "08-从能跑到能稳定跑很久——爬虫的工程化跃迁"
created: 2026-05-06
tags:
  - 项目
aliases:
  - 第 5 讲：从'能跑'到'能稳定跑很久' —— 爬虫的工程化跃迁
---

# 从能跑到能稳定跑很久——爬虫的工程化跃迁

前几讲,已经能写出一个会发请求、会解析、会绕反爬的爬虫了。但**它还只是个"脚本"**——点一下运行,它跑完就结束。

而真实项目里的爬虫要做的事完全不一样:

- **24 小时不停跑**,有时候要连续跑几个月
- 要抓**几百家公司、几十万个职位**
- 中途**网络断了、电脑重启了、被封了**,得能自己恢复
- 抓到的数据要**进数据库**、要**去重**、要**追踪状态**
- 已经入库的数据**过期了要清洗**、**链接失效了要回写**

这些事情**不是写代码能力的问题,而是工程设计的问题**。

这一讲需要来回答最后一个核心问题:**怎么把一个会跑的脚本,变成一个能稳定跑很久的系统?**

学完这一讲,回头看猎聘项目里那个 1000 行的 `spider_liepin_company.py`,你会发现它其实没什么神秘的——它只是把这一讲讲的几件事,**每一件都做到了"豪华版"**而已。

## **一、从脚本到系统:核心矛盾在哪里?**

把脚本变成系统,会遇到 5 个绕不开的问题。

一个一个解决。

```text
问题1:配置硬编码 → 换个站点要改一堆代码    → 配置驱动
问题2:跑到一半崩了 → 重新跑一遍前面白做     → 状态系统
问题3:抓的数据越来越多 → 单进程跑不完       → 任务切片 + 多实例
问题4:数据怎么用?怎么去重? → 不能堆在硬盘  → 入库链路
问题5:已入库数据会过期、失效 → 数据库会脏   → 清洗回写
```

这 5 个问题对应着工程化爬虫的 5 个核心模块。逐个拆开讲。

---

## **二、问题 1:配置驱动 —— 把"会变的"和"不变的"分开**

我们在第 4 讲写的爬虫,所有 URL、字段名、目录路径都**直接写死在代码里**:

```properties
URL = "https://realpython.github.io/fake-jobs/"
DATA_DIR = "data/jobs"
```

这个写法的问题:换一个站点要抓,**整个代码都要改**。换 10 个站点,意味着要写 10 份几乎一样的代码。

**正确做法:把"会变的"抽出来放到配置文件里。**

#### **2.1 用 INI 配置文件**

```properties
# 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
```

代码里:

```python
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 家公司"、"我要切到生产环境跑一下"。这种**临时性指令**用命令行参数:

```python
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)
```

启动:

```text
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**

更进阶的做法是**配置叠加**:

```text
setting_default.ini         ← 全局默认值
  ↓ 被覆盖
setting_sch_99.ini         ← 第 99 个分片的特定配置
  ↓ 被覆盖
命令行参数 -d prod          ← 临时覆盖
```

读取顺序:先读默认,再读特定分片,最后命令行覆盖。这样**既能保持默认值统一,又能让每个分片有自己的个性化设置**。通用结构化系统采用的就是这套层级。

## **三、问题 2:状态系统 —— 让爬虫"记得自己抓到哪了"**

第 4 讲我们用 `progress.txt` 记录"抓到第几条"。但真实项目里,事情远比这复杂:

- 一个公司有上千个职位,要按"地区 × 经验 × 薪资"切成上百个组合,**每个组合都得记录是否抓完**
- 程序中途崩了,**状态文件不能损坏**(否则全完了)
- 老格式的状态文件要**平滑升级**到新格式,不能丢数据

这就是猎聘公司版那套 `spider_status_<ComId>.json` 解决的问题。

### 3.1 状态文件的本质:就是个 JSON

```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 关键操作:判断 / 标记 / 完成率**

```python
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
```

主流程:

```text
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()` 有个隐患:**写到一半程序崩了,文件就坏了**。下次启动读不出来,所有状态丢失。

工程化做法:**先写临时文件,再原子重命名**。

```python
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 格式兼容与迁移**

项目跑久了,会想升级状态文件格式(比如从字典换成数组以提高性能)。但**线上几千个老文件不能丢**。

```python
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 条**(深翻页都受限制)。怎么办?

**把"大查询"切成"小查询"**——按地区、经验、薪资、职类做笛卡尔积:

```java
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 个组合就是浪费——每个组合都要发请求、都要落盘。**做个阈值判断**:

```properties
total = get_total_count(company)

if total <= 120:
    # 直接抓一轮
    crawl_simple(company)
else:
    # 切片
    for cond in generate_conditions(company):
        crawl_one(cond)
```

**这就是猎聘代码里那个 "120 阈值" 的由来**——基于实际数据观察出的拐点。

### **4.3 多实例分片:把任务分给多台机器**

更大的体量,要靠**多机并行**:

```text
# 机器 1
python main.py -f 1   # 跑分片 1(公司 ID 模 10 == 1)

# 机器 2
python main.py -f 2

# 机器 10
python main.py -f 10
```

代码里:

```python
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 小时不停打**

```python
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 小时跑?

1. **凌晨流量小**,你抓得猛容易被识别
2. **抓久了代理消耗大**,适当休息能让代理池补货
3. **业务有节奏**——招聘网站本身上班时间发布的新职位多,夜里没多少新数据,抓也是浪费

## **五、问题 4:入库链路 —— "先文件、后接口"**

抓到数据之后,**不要直接往数据库写**。这是工程化爬虫最重要的一条原则。

### **5.1 为什么不直接入库?**

```yaml
直接入库:
爬虫 → 数据库
    ↑
   这一步出错,数据就没了

先文件后接口:
爬虫 → 本地 JSON → API → 数据库
       ↑           ↑
       数据保住了   API 失败可以重试
```

**好处:**

1. **可追溯**:出问题了,能看到每条数据当时长什么样
2. **可重跑**:解析逻辑改了,直接重跑本地 JSON,不用重新抓网络
3. **解耦**:抓取和入库是两件事,各自可以独立扩展、独立调试
4. **批量重试**:某个 API 挂了 1 小时,这 1 小时的文件等 API 恢复了批量重传

### **5.2 完整链路:落盘 → 调接口 → 归档**

```python
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 目录就是状态机

```text
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 不只返回"成功/失败",还可能返回**业务码**:

```properties
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 清洗爬虫的特殊性**

| 普通采集爬虫 | 清洗爬虫 |
| --- | --- |
| 目标:抓新数据 | 目标:验证老数据 |
| 字段:抓全 | 字段:不抓,只看页面状态 |
| 数据流:站点 → 数据库 | 数据流:数据库 → 站点 → 数据库(回写状态) |
| 规模:百万级 | 规模:千万级(存量更大) |

清洗爬虫的输入是数据库已有的链接列表,输出是过期标记。

```python
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` 加了一堆优化:

1. **浏览器上下文复用**:不为每个链接开新 context,而是复用同一个,**节省启动开销**
2. **LRU 缓存**:同一批次里如果有重复 URL,直接用缓存结果
3. **指数退避重试**:网络抖动时重试,失败间隔逐步拉长
4. **多选择器兼容**:验证页可能有多种样式,用一组选择器轮询匹配
5. **统一资源清理**:用 `with` 和上下文管理器保证浏览器实例最终一定被关掉

```python
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)
```

**这就是项目里"工程化程度最高的脚本"长什么样**——它不是逻辑最复杂,而是**对性能、稳定性、资源管理的考虑最周密**。

---

## **七、问题之外:基类抽象 —— 把"共性"集中起来**

到这里你应该发现,无论是哪个站点的爬虫,都要做同样的事:**读配置、获取代理、判断去重、调入库接口、归档文件、记录日志**。

如果每个站点脚本都重写一遍,**维护成本爆炸**。所以要抽出一个**公共基类**:

```python
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()`,所有子类自动生效。这就是为什么真实项目里基类那么重要。

---

## **八、把所有模块串起来:工程化爬虫的全貌**

讲到这里,我们可以把一个工程化爬虫的完整架构画出来了:

```text
                  ┌──────────────────────────────┐
                  │     配置层 (.ini + CLI)       │
                  │  路径、接口、参数、模式       │
                  └────────────┬─────────────────┘
                               │
                  ┌────────────▼─────────────────┐
                  │     调度层 (main.py)          │
                  │  按模式分发、按分片切分       │
                  └────────────┬─────────────────┘
                               │
                  ┌────────────▼─────────────────┐
                  │     基类层 (spider.py)        │
                  │  代理 / 去重 / 入库 / 归档    │
                  └────┬──────────┬──────────┬───┘
                       │          │          │
              ┌────────▼──┐ ┌─────▼────┐ ┌───▼────┐
              │ LiepinSp. │ │ YupaoSp. │ │ 51JobSp│
              │  (子类)    │ │ (子类)    │ │ (子类) │
              └────────┬──┘ └─────┬────┘ └───┬────┘
                       │          │          │
                       └─────┬────┴──────────┘
                             ▼
                  ┌──────────────────────────────┐
                  │     落盘层 (data/)            │
                  │  job/ → success/ / failed/   │
                  └────────────┬─────────────────┘
                               │
                  ┌────────────▼─────────────────┐
                  │     入库层 (内部 API)         │
                  │  jobparse / comparse         │
                  └────────────┬─────────────────┘
                               │
                  ┌────────────▼─────────────────┐
                  │     清洗层                    │
                  │  yupao_clean / job_to_expired │
                  └──────────────────────────────┘
```

**这就是猎聘 + 通用结构化系统的完整骨架**。每一层都解决了一个特定问题,合起来就是"能稳定跑很久"。

---

## **九、本讲核心认知**

> **"工程化"不是把代码写得复杂,而是把'每一种会出错的情况'都提前安排好**

| 会出什么错 | 怎么预防 |
| --- | --- |
| 配置改了要重新发版 | 配置文件 + 命令行参数 |
| 跑一半崩了 | 进度文件 + 状态文件 + 原子写入 |
| 单机跑不动 | 分片 + 多实例 |
| 数据丢了 | 先文件后接口 + 三态归档 |
| 数据库变脏 | 清洗爬虫 + 回写过期 |
| 改一处要改十处 | 基类抽象 |
| 站点改版了 | 配置驱动 + 模板化抽取函数 |

回头去看 `spider.py` 的初始化代码,你会发现它不过是在做"读配置、建目录、初始化代理客户端、加载状态文件"——**没有任何花哨的东西,只是把所有该做的事都做了**。

---

**组内导航**：⬅️ [[07-练手网站|练手网站]] | 🏠 [[00-爬虫课程六讲|00-爬虫课程六讲]] | ➡️ [[09-对照项目源码拆解——从配置到入库的完整链路|对照项目源码拆解——从配置到入库的完整链路]]