★ FEATURED ARTICLE 做了几年数据采集相关的工作我踩过最多的坑不是反爬而是爬虫跑到一半挂了几十万条数据眼睁睁没了一半。更头疼的是重启之后不知道从哪里继续爬只能重头再来。分页存储和断点续爬这两个设计就是专门解决这类问题的。之前我维护过一套商品数据采集任务计划抓几十万条记录跑到差不多三分之二的时候服务器一次内存溢出直接把进程带走了。重启之后面临的问题非常现实日志丢了、内存队列清空了、不知道哪些页面已经抓过只能把整个任务推倒重来。那一次之后我把分页存储和断点续爬这两件事列为了爬虫架构的基础设施之后所有采集任务都按这套思路来做。这篇文章就把这套方案的原理和落地过程完整梳理一遍适合那些已经写过简单爬虫、开始认真考虑数据组织和任务可靠性的开发者参考。1. 先从“为什么”说起分页存储与断点续爬到底解决什么问题1.1 一次性全量入库的隐性风险很多刚接触爬虫的开发者习惯这样写把所有结果先塞进一个内存列表等任务跑完了再一次写入数据库。代码看起来简单但一旦数据量上来问题就接踵而至。第一个问题是内存。假设一条记录平均占 2KB 的有效字段十万条就是 200MB内存里再叠加 request 对象、解析中间结果、去重集合轻轻松松吃掉一个 2G 的云服务器内存。我在实际项目中见过最夸张的情况一个任务进程内存涨到 4G系统直接触发 OOM kill。第二个问题是数据写入时机太晚。如果任务在最后一步“写库”之前崩溃前面几个小时抓到的所有有效数据全部消失。爬虫本身在高并发、网络不稳定、目标站点变化的环境下运行崩溃几乎是必然事件问题只在于你允许多大的数据损失。第三个问题是查询和后续处理的效率。一次性写入一个大表后续如果要按批次、按日期、按来源分析数据没有清晰的物理或逻辑分区每一条查询都要全表扫描越到后面越痛苦。1.2 断点续爬的核心价值断点续爬解决的是“任务中断后如何恢复”的问题。它的本质是把任务进度状态持久化让爬虫在重启之后能够精确地从中断位置继续而不是从头再来。听起来简单但要做到“精确”并不容易。你至少需要回答三个问题哪些页面已经抓取完成了哪些页面抓取了但失败了需要重试当前处理到整个任务队列的哪个位置这三个问题不解决所谓“续爬”就只是在碰运气。很多人的做法是把 URL 列表存下来重启后从第一个 URL 重新跑一遍再用去重逻辑跳过已抓取的页面。这个策略在数据量小的时候勉强能用但数据量一大光是逐条比对去重集合就会拖垮性能更不用说去重集合本身也是内存大户。1.3 适用的场景边界分页存储加断点续爬不是万能的它最适合以下几类场景大规模采集任务单个任务预期产生几万条以上数据长时间运行任务任务耗时超过 10 分钟存在中断风险增量更新任务每天定时爬取新数据需要避免重复采集多批次并行任务多个爬虫实例分工协作需要统一管理进度反过来如果你只是临时抓几百条数据一次内存列表加一次写库就够了没必要引入额外的复杂设计。工程上的稳定性和复杂度是成正比的关键在于判断什么时候值得付出这个复杂度。2. 分页存储设计从内存到落盘的完整链路2.1 先想清楚分页的粒度分页存储的“页”到底指什么这个粒度决定了整套设计的走向。我在实际项目中通常把分页分成三个层级源站分页目标站点本身的分页参数第 1 页、第 2 页…抓取批次分页每次循环批量抓取的一组 URL存储分页数据文件或数据库表按某种规则切块其中最关键的是存储分页。源站分页是对方定义的你只能去适配而存储分页是你自己的数据组织方式完全由你来设计。存储分页做得好查询、备份、增量更新都会非常顺手。常见的存储分页方案有三种按时间分片、按 ID 范围分片、按固定行数分片。三种方案各有适用场景没有绝对的好坏。2.2 三种分页方案的对比与选型分页方案实现思路优点缺点适合场景按时间分片每天/每小时一个存储分片查询按时间走天然支持增量单日数据量大时仍会膨胀日志类、按日期统计的数据按 ID 范围分片主键落在某个区间就存到对应分片分布均匀查询可通过主键定位需要生成全局有序 ID用户行为、订单等流水数据按固定行数分片每 N 条记录切割一次实现简单分片大小可控无法按业务维度查询定位通用采集结果存储我给商品采集项目选的是主键 ID 加固定行数混合方案。原因很直接采集结果没有强时间属性同一批商品入库时间可能分散按时间分片会导致分片大小不均而固定行数分片可以保证每个分片文件的大小稳定方便后续迁移和归档。提示选分页方案前先想清楚下游要拿这些数据做什么。如果下游是做报表的按时间分片一定优先如果下游是做数据仓库 ETL 的按固定行数分片配合列式存储会更友好。方案是跟着使用场景走的不要为了“设计感”强行选一个。2.3 基于 SQLite 的分页落盘实现SQLite 是中小型爬虫项目最实用的存储方案单文件、零配置、支持事务比直接写 JSON 文件可靠得多又比上 MySQL/PostgreSQL 省去运维成本。实现分页存储的逻辑是为每个分片创建一个独立的 SQLite 数据库文件文件名带上分片索引。import sqlite3 import os from datetime import datetime class PaginatedStorage: def __init__(self, base_dir, rows_per_page10000): self.base_dir base_dir self.rows_per_page rows_per_page os.makedirs(base_dir, exist_okTrue) self.current_page None self.current_conn None self.current_count 0 self.meta_conn self._connect_meta() def _connect_meta(self): conn sqlite3.connect(os.path.join(self.base_dir, meta.db)) conn.execute( CREATE TABLE IF NOT EXISTS page_index ( page_id INTEGER PRIMARY KEY AUTOINCREMENT, file_name TEXT NOT NULL, created_at TEXT NOT NULL, record_count INTEGER DEFAULT 0, status TEXT DEFAULT active ) ) conn.commit() return conn def _next_page_id(self): cursor self.meta_conn.execute( SELECT COALESCE(MAX(page_id), 0) 1 FROM page_index ) return cursor.fetchone()[0] def _open_new_page(self): page_id self._next_page_id() file_name fpage_{page_id:06d}.db conn sqlite3.connect(os.path.join(self.base_dir, file_name)) conn.execute( CREATE TABLE IF NOT EXISTS items ( id INTEGER PRIMARY KEY AUTOINCREMENT, source_url TEXT NOT NULL, title TEXT, content TEXT, crawled_at TEXT NOT NULL ) ) conn.commit() self.meta_conn.execute( INSERT INTO page_index (file_name, created_at) VALUES (?, ?), (file_name, datetime.now().isoformat()) ) self.meta_conn.commit() self.current_page page_id self.current_conn conn self.current_count 0 return conn def write_item(self, item): if (self.current_conn is None or self.current_count self.rows_per_page): self._open_new_page() self.current_conn.execute( INSERT INTO items (source_url, title, content, crawled_at) VALUES (?, ?, ?, ?), (item[source_url], item[title], item[content], datetime.now().isoformat()) ) self.current_conn.commit() self.current_count 1 if self.current_count 1: # 第一个元素写入时更新元信息 self.meta_conn.execute( UPDATE page_index SET record_count record_count 1 WHERE page_id ?, (self.current_page,) ) self.meta_conn.commit() def close(self): if self.current_conn: self.current_conn.commit() self.current_conn.close() self.meta_conn.commit() self.meta_conn.close()这个实现有几个设计细节值得说明。每一次写满rows_per_page条就切换到新的分片文件切片的动作发生在write_item方法的入口处而不是写入之后这样能避免一次性多写几条超出分片容量的问题。分片索引表page_index独立放在 meta.db 中相当于整个存储的索引目录后续做数据汇总、统计分片数量、清理旧分片都非常方便。还有一个容易被忽视的点每个分片内部必须做commit。SQLite 默认在事务中不会自动落盘如果不主动 commit进程崩溃时最后一批数据会全部丢失。这一点在爬虫场景下尤其致命因为你可能已经向源站发出了几百个请求结果数据没写稳。2.4 文件型分页存储的取舍有些场景不一定要用数据库直接用文件分页也是合理的。比如你抓的是图片、PDF 等二进制文件或者你想把原始响应报文存档以备分析这时用 JSON Lines 按行追加写入比数据库更合适。JSON Lines 的核心好处是每行一个独立 JSON 对象追加写入天然支持分页只要按约定在某个行数或某个时间点切换新文件即可。而且格式非常灵活字段增减不需要改表结构对快速演进的采集任务非常友好。import json class JSONLPaginatedWriter: def __init__(self, base_dir, lines_per_file5000): self.base_dir base_dir self.lines_per_file lines_per_file self.current_file None self.current_lines 0 def _open_new_file(self): file_index self.current_lines_per_file_count() file_path f{self.base_dir}/data_{file_index:05d}.jsonl self.current_file open(file_path, a, encodingutf-8) self.current_lines 0 def write_record(self, record): if self.current_file is None or self.current_lines self.lines_per_file: self._open_new_file() self.current_file.write(json.dumps(record, ensure_asciiFalse) \n) self.current_lines 1用 JSONL 的真实好处是下游做数据处理的时候不需要解析大 JSON 数组直接按行读即可内存占用是 O(1) 的。而用一个大 JSON 数组存十万条记录数据解析时要把整个数组加载到内存和小文件分片相比完全不是一个体验。不过文件方案也有明显短板不支持去重、不支持按条件查询、并发写入容易出问题。所以我的建议是文本类结构化数据优先用 SQLite 分片原始报文和二进制文件用 JSONL/文件分片数据量大到单机 SQLite 撑不住时再考虑迁移到 MySQL 的分库分表方案。3. 断点续爬实现状态记录与恢复机制3.1 断点续爬的本质是记录“进度状态”断点续爬听起来玄乎本质就是一个状态机问题。爬虫任务的状态可以抽象为待抓取、抓取中、已成功、已失败。断点续爬做的事情不过是把这个状态持久化到磁盘让进程重启后能恢复这个状态机。这里有一个最容易犯的错误把“进度”和“数据”混在一起。很多人把已经抓到的数据当成进度认为“只要数据落盘了重启后去重跳过就行”。这在数据量小时没问题但数据量大了之后去重成本会指数上升。更隐蔽的问题是有些页面抓取成功但解析失败或者 HTTP 请求超时但数据其实已部分返回这些“中间态”没有单独记录重启后你会完全失去判断依据。正确的做法是把任务进度单独维护用一张状态表记录每一个 URL 的完整生命周期。数据落盘是数据层的事情状态推进是任务层的事情两层严格分离。3.2 状态表设计与检查点写入策略状态表的设计直接决定断点续爬的上限。我在项目中通常使用如下的表结构CREATE TABLE task_state ( url TEXT PRIMARY KEY, page_index INTEGER NOT NULL, status TEXT NOT NULL, -- pending / fetching / done / failed retry_count INTEGER DEFAULT 0, last_error TEXT, updated_at TEXT NOT NULL ); CREATE INDEX idx_status ON task_state(status); CREATE INDEX idx_page ON task_state(page_index);这里的page_index字段是分页存储和断点续爬之间的桥梁它记录当前 URL 的数据应该归到哪个分片。状态表用 URL 作为主键天然具备去重能力同时status加上retry_count可以精确控制重试逻辑。检查点的写入策略同样重要。最朴素的方案是每处理完一个 URL 就 update 一次状态但高频写入会严重拖慢速度。我在实践中采用的是“批量检查点 定期 flush”的组合策略每处理完 100 个 URL提交一次状态更新事务每完成一个分页的写入同步提交一次索引元信息进程收到终止信号时立刻 flush 当前批次的状态这个策略既保证了进度的实时性又不至于因为频繁提交反而拖垮性能。关键原则是状态更新和数据写入必须是一个原子操作要么都成功要么都失败。import sqlite3 from contextlib import contextmanager class CheckpointManager: def __init__(self, db_path, flush_interval100): self.conn sqlite3.connect(db_path) self.conn.execute( CREATE TABLE IF NOT EXISTS task_state ( url TEXT PRIMARY KEY, page_index INTEGER NOT NULL, status TEXT NOT NULL, retry_count INTEGER DEFAULT 0, last_error TEXT, updated_at TEXT NOT NULL ) ) self.conn.execute(CREATE INDEX IF NOT EXISTS idx_status ON task_state(status)) self.conn.commit() self.flush_interval flush_interval self.pending_updates [] self.pending_count 0 contextmanager def transaction(self): try: yield self.conn.commit() except Exception: self.conn.rollback() raise def mark_pending(self, url, page_index): self.pending_updates.append( (url, page_index, pending, 0, None, self._now()) ) def mark_success(self, url, page_index): self.pending_count 1 self.pending_updates.append( (url, page_index, done, 0, None, self._now()) ) if self.pending_count self.flush_interval: self.flush() def mark_failed(self, url, page_index, error_msg): self.pending_count 1 self.pending_updates.append( (url, page_index, failed, self._get_retry_count(url) 1, error_msg, self._now()) ) if self.pending_count self.flush_interval: self.flush() def flush(self): if not self.pending_updates: return with self.transaction(): for url, page_index, status, retry, error, ts in self.pending_updates: self.conn.execute( INSERT INTO task_state (url, page_index, status, retry_count, last_error, updated_at) VALUES (?, ?, ?, ?, ?, ?) ON CONFLICT(url) DO UPDATE SET statusexcluded.status, retry_countexcluded.retry_count, last_errorexcluded.last_error, updated_atexcluded.updated_at, page_indexexcluded.page_index, (url, page_index, status, retry, error, ts) ) self.pending_updates [] self.pending_count 0 def recover_remaining(self): cursor self.conn.execute( SELECT url, page_index FROM task_state WHERE status ! done ORDER BY retry_count ASC ) return cursor.fetchall() def _now(self): from datetime import datetime return datetime.now().isoformat() def _get_retry_count(self, url): cursor self.conn.execute( SELECT retry_count FROM task_state WHERE url ?, (url,) ) row cursor.fetchone() return row[0] if row else 0这里用ON CONFLICT(url) DO UPDATE的 upsert 写法可以很自然地处理同一个 URL 多次更新的情况不需要先查询再更新。3.3 恢复流程与幂等设计恢复流程的核心是把recover_remaining()返回的未完成任务重新放进队列。但这里有一个非常隐蔽的问题status标记为done的数据一定完整吗不一定。有可能页面已经抓取成功数据也写入了分片文件但状态更新却因为进程崩溃没来得及提交。重启后这个 URL 会重新进入待抓取队列然后产生重复数据。解决这个问题的思路有两个方向。第一个是“先更新状态再写数据”这样崩溃时数据缺失但状态已经是 done恢复逻辑会跳过它——但这会让数据永久缺失。第二个是“先写数据再更新状态”崩溃时数据在但状态不是 done恢复后会重复抓取产生重复数据。两种方案都有缺陷。我的选择是第二种再加一层幂等保护在数据表里给source_url加唯一索引重复写入时采用INSERT OR IGNORE或先查重再插入。这样既不会丢失数据也不会产生重复。def process_url_with_idempotency(url, page_index, storage, checkpoint): response fetch_url(url) try: items parse_response(response) storage.write_item({source_url: url, **items}) checkpoint.mark_success(url, page_index) except Exception as exc: checkpoint.mark_failed(url, page_index, str(exc))配合唯一索引conn.execute( CREATE UNIQUE INDEX IF NOT EXISTS idx_url_unique ON items(source_url) )有了这个唯一索引即使同一个 URL 被重复处理写入时也会被数据库拦下来。数据层的幂等保护是断点续爬的最后一道防线状态逻辑上的小漏洞可以被它兜住。3.4 失败重试与去重策略断点续爬还有一个容易被低估的点失败重试策略。网络请求失败是常态但不意味着每个失败都要立刻重试。我在项目中按照错误类型做了分级处理错误类型处理方式重试次数上限重试间隔超时错误直接重试3指数退避初始 1 秒HTTP 5xx等几秒后重试3固定 5 秒HTTP 4xx不重试记录错误0无解析错误重试 1 次1立即4xx 错误基本是固定的比如 404 页面不存在、403 无权限重试多少次结果都一样只会浪费资源。解析错误有可能是临时性的比如目标站点改版了 HTML 结构重试一次看看是运气问题还是结构变化。重试间隔采用指数退避是必须的否则同一个时间点所有失败的请求同时重试会给目标服务器造成瞬时压力也容易触发更严格的反爬限制。4. 实操过程一个完整的落地示例4.1 环境准备与整体架构为了把上面的设计串起来我用一个完整的示例来演示从零到一落地一套“分页存储 断点续爬”的采集任务。示例的目标是一个模拟的商品列表数据源我把它命名为模拟数据接口只用于本地演示不涉及任何真实站点。整体架构如下任务管理器负责生成待抓取的 URL 列表并维护进度状态抓取器负责网络请求和页面解析存储层负责按分片写入 SQLite检查点管理器负责持久化任务状态。四层各司其职通过明确的接口协作。pip install requests beautifulsoup4需要用到的库就这两个。SQLite 是 Python 标准库自带的不需要额外安装。整个项目 200 行代码以内就能跑通这也是我愿意推荐这种方案的原因——足够简单足够可靠。4.2 任务管理器的实现先定义一个任务管理器负责初始化和恢复状态。import requests from bs4 import BeautifulSoup from storage import PaginatedStorage from checkpoint import CheckpointManager class CrawlerTask: def __init__(self, base_dir, total_pages500): self.storage PaginatedStorage(f{base_dir}/data) self.checkpoint CheckpointManager(f{base_dir}/task_state.db) self.total_pages total_pages self.base_url https://example.local/list?page{} def _generate_url_list(self): 生成待抓取 URL 列表并注册到状态表 for page in range(1, self.total_pages 1): url self.base_url.format(page) self.checkpoint.mark_pending(url, page) self.checkpoint.flush() def run(self): remaining self.checkpoint.recover_remaining() if remaining: print(f检测到未完成任务{len(remaining)} 条从断点恢复) else: print(首次运行初始化任务队列) self._generate_url_list() remaining self.checkpoint.recover_remaining() processed 0 for url, page_index in remaining: success self._crawl_single(url, page_index) if success: processed 1 if processed % 100 0: print(f已完成 {processed} 条当前分片 f{self.storage.current_page})这里有一个关键点每次run()的第一步永远是recover_remaining()而不是重新生成 URL 列表。这样无论任务跑了多少次、中断了多少次都会被引导到正确的恢复路径上。4.3 抓取循环与数据入库抓取单个 URL 的流程包含请求、解析、写入、更新状态四个步骤。import time import random def _crawl_single(self, url, page_index): max_retries 3 for attempt in range(max_retries): try: resp requests.get(url, timeout10) resp.raise_for_status() soup BeautifulSoup(resp.text, html.parser) items [] for node in soup.select(.product-item): title node.select_one(.title).get_text(stripTrue) price node.select_one(.price).get_text(stripTrue) items.append({title: title, price: price}) if not items: # 页面结构可能变了标记失败等待人工处理 self.checkpoint.mark_failed(url, page_index, empty parse result) return False for item in items: self.storage.write_item({ source_url: url, title: item[title], price: item[price], }) self.checkpoint.mark_success(url, page_index) return True except requests.Timeout: sleep_time 2 ** attempt random.uniform(0, 1) time.sleep(sleep_time) except requests.HTTPError as exc: if exc.response.status_code 404: self.checkpoint.mark_failed(url, page_index, 404 not found) return False if exc.response.status_code 500: time.sleep(5) else: self.checkpoint.mark_failed(url, page_index, str(exc)) return False self.checkpoint.mark_failed(url, page_index, max retries exceeded) return False注意这里的失败处理逻辑4xx 错误直接标记失败不重试5xx 错误等待 5 秒重试超时错误用指数退避。每次失败后必须显式调用mark_failed并 flush否则重启后会丢失这些失败记录。还有一个不能省的细节每个页面抓取成功后要主动控制请求频率。示例中简单地把页面解析挂在请求后实际上线时我会在循环里加一个平均间隔控制比如每 3 秒一个请求再叠加随机的抖动时间避免请求节奏过于规律。4.4 运行效果与控制台输出跑起来之后控制台输出大致会是这样的首次运行初始化任务队列 检测到未完成任务500 条从断点恢复 已完成 100 条当前分片page_000002.db 已完成 200 条当前分片page_000002.db ... 进程在这里被手动终止模拟崩溃...重启后再次执行检测到未完成任务187 条从断点恢复 已完成 300 条当前分片page_000004.db看到“检测到未完成任务187 条”这句话说明断点续爬生效了。之前已经完成的部分数据落在 page_000001.db 到 page_000003.db 三个分片里通过 meta.db 的 page_index 表可以随时统计总量和分片明细。验证数据完整性时我通常直接查 meta.dbSELECT page_id, file_name, record_count FROM page_index ORDER BY page_id;输出1 | page_000001.db | 10000 2 | page_000002.db | 10000 3 | page_000003.db | 4802如果你看到最后一个分片的 record_count 小于分片容量不用慌那只是任务中断时的正常状态续爬后会继续填满它或者开启新分片。5. 常见问题与排查技巧实录5.1 数据丢失写库和更新状态到底谁先谁后这是断点续爬项目里我被问到最多的问题。我的答案在前面已经明确先写数据后更新状态再靠唯一索引去重兜底。但这里有一个细节值得展开。你写数据用的是分片存储的write_item更新状态用的是检查点管理的mark_success这两个操作在物理上不是同一个数据库。如果两个操作中间进程崩溃就会出现“数据在状态未更新”的中间态。处理方式有两种。第一种是容忍它靠唯一索引切掉重复数据第二种是做“两阶段提交”先写入一个待确认标记数据完全落盘后再确认状态。考虑到爬虫项目的时效性第一种方式完全够用。我在线上跑了很久重复数据率始终是 0靠的就是数据表的唯一索引。注意设置唯一索引的前提是你的数据表里有source_url这类天然唯一的字段。如果只有标题、时间这种不唯一的信息幂等保护就无从谈起就要回到两阶段提交的方案。5.2 恢复后重复爬取了一大片页面重启后发现恢复的 URL 数量明显偏多或者很多 URL 在抓取时被唯一索引拦截。这个现象通常指向两个原因状态表提交太慢或者 flush 间隔设置得太长。检查状态更新的flush_interval如果设置成 1000那么最后一个批次里最多有 999 个已完成 URL 没来得及提交。进程恰好在这时崩溃这 999 个 URL 全部会重新进入队列。解决思路很简单缩小 flush 间隔比如调到 100或者在任务捕获到终止信号时强制调用一次flush()。你要清楚一个权衡flush 越频繁状态越实时但写入开销越大。100 是一个在绝大多数场景下都能接受的平衡点。5.3 并发场景下的状态冲突如果你用多线程或多进程并行抓取状态表会面临并发写入冲突。SQLite 默认只允许一个写者多个进程同时写状态表会导致database is locked。针对多线程最简单的做法是在CheckpointManager内部加一个线程锁让所有线程共享同一个连接对象串行化状态写入。状态写入本身开销极低串行化不会成为性能瓶颈。针对多进程建议给每个进程分配独立的 URL 子集各自维护独立的状态库文件最后再合并。不要试图让多个进程同时写同一个 SQLite 文件那不是 SQLite 的适用场景。任务量大到单机多进程都撑不住时再考虑换 MySQL 或者引入消息队列。5.4 分片文件无限增长查询性能下降分页存储解决了一开始的数据膨胀问题但时间长了分片文件也会越来越多。我遇到过的一个实际问题是分片数量上千个之后对 meta.db 的索引查询变慢对整个数据集做统计要遍历所有分片文件。我的处理策略是增加一个“归档”机制定期把超过保留期的旧分片压缩归档移出热目录。同时如果你确实需要对全量数据做分析考虑在数据仓库层建立一个汇总表分片数据在写入时同步往汇总表里打一份。另外一个容易忽略的性能优化点每个分片文件创建索引。只在分片建表时创建一次不要在每次写入时重复维护索引结构。def _open_new_page(self): # 建表之后立刻创建索引 self.current_conn.execute( CREATE INDEX IF NOT EXISTS idx_crawled_at ON items(crawled_at) )5.5 反爬策略影响下的常见误判从实际经验看很多“断点续爬失效”的案例根因反而是被反爬机制干扰了。比如请求太频繁触发验证码页面返回的不是预期结构而是验证页面解析出的 items 为空被误判为“页面结构变更”实际上只是请求频率问题。排查思路在解析结果为空时先把原始响应前 200 个字符打印到日志里快速判断是结构问题还是反爬问题。同时重试逻辑里对“空解析结果”不要直接判死可以加重试次数但每次重试间隔拉得长一点。我习惯的做法是空结果连续出现 3 次才标记失败给临时性策略留出恢复时间。还有一个经验分页存储的文件名里带上本次任务开始的时间戳比如page_20250115_000001.db。这样如果某个批次的数据出现异常你能快速定位到是哪一天、哪一个任务的数据。6. 实操总结整套方案跑下来我最大的体会有两点。第一点是状态与数据分离。很多人把“数据写进了数据库”当成“任务进度完成了”这个认知在高并发、长耗时任务里会造成很严重的误判。数据落盘只代表采集结果有了着落不代表任务推进到了某个阶段。只有把 URL 生命周期状态单独维护起来才能真正做到精确恢复。第二点是幂等设计永远不嫌早。哪怕你现在只是写个小脚本我也建议在数据表里加上 URL 唯一索引理由很简单加了索引之后无论任务逻辑怎么改、怎么重启、怎么出现“重复处理”数据库都会安静地帮你挡住重复数据。这个成本极低收益却极高。最后分享一个小技巧。我在每次任务结束时都会在 meta.db 里记录一条任务汇总记录内容包括开始时间、结束时间、总请求数、成功数、失败数、分片数量。虽然看起来只是多写了几个字段但当你需要评审数据质量、排查历史问题时这条记录能省去大量回溯时间。真正稳定的爬虫架构不是靠某一个大招而是靠这些细小的习惯堆出来的。
阅读完成 · 觉得有帮助？
