在电商数据采集场景中，爬虫常因网络波动、服务器重启、IP 封禁等问题中断。若缺乏断点续爬机制，重启后需从头开始，不仅浪费带宽与时间，还可能因重复采集导致数据冗余。Redis 凭借其高性能、原子操作、多样数据结构的特性，成为实现断点续爬的理想工具。本文将从核心原理到实战代码，完整讲解实现方案。

一、为什么选择 Redis 实现断点续爬？

断点续爬的核心需求是记录 “已爬状态” 和 “待爬任务”，确保中断后能从上次进度继续。Redis 的优势恰好匹配这些需求：

1. 高性能读写：电商爬虫 URL 队列常达百万级，Redis 每秒 10 万 + 的读写速度可支撑高并发任务调度；
2. 原子操作保障：分布式爬虫多实例抢任务时，Redis 的LPOP/SPOP等原子命令可避免重复取 URL；
3. 多样数据结构：

• • 待爬 URL 队列：用List（FIFO）或Sorted Set（按优先级排序）；

• • 已爬 URL 记录：用Set（O (1) 判断是否重复，自动去重）；

• • 临时数据缓存：用Hash存储已采集但未入库的数据，防止爬后丢数据；
    
    4.持久化机制：支持 RDB/AOF 持久化，即使 Redis 重启，断点数据也不会丢失。

二、核心实现逻辑：三大核心模块

断点续爬的本质是 “状态记录” 与 “进度恢复”，需围绕以下三个模块设计 Redis 存储方案：

1. 待爬 URL 队列（任务池）

作用：存储尚未采集的电商 URL（如商品列表页、详情页），确保爬虫能持续获取任务。

数据结构选择：

• 简单场景（无优先级）：用List，通过LPUSH（添加任务）和RPOP（获取任务）实现 FIFO 队列；

• 复杂场景（有优先级）：用Sorted Set，为不同 URL 设置分数（如 “商品详情页 = 10，分类列表页 = 5”），通过ZADD添加任务，ZRANGEBYSCORE按优先级取任务。

示例命令（List 实现）：

# 添加待爬URL（如电商商品详情页）
LPUSH e_commerce:wait_urls "https://xxx.com/item/123"
LPUSH e_commerce:wait_urls "https://xxx.com/item/456"# 取出待爬URL（爬虫消费任务）
RPOP e_commerce:wait_urls  # 返回"https://xxx.com/item/123"

2. 已爬 URL 记录（去重池）

作用：记录已采集完成的 URL，避免重复爬取（电商 URL 重复爬会导致数据冗余，还可能触发反爬）。

数据结构选择：Set（天然去重，判断 URL 是否已爬的时间复杂度为 O (1)）。

核心逻辑：

1. 爬虫获取待爬 URL 后，先通过SISMEMBER判断是否在已爬集合中；
2. 若已存在，直接跳过；若不存在，执行采集逻辑；
3. 采集成功后，用SADD将 URL 加入已爬集合。

示例命令：

# 判断URL是否已爬
SISMEMBER e_commerce:done_urls "https://xxx.com/item/123"  # 返回0（未爬）# 采集成功后，加入已爬集合
SADD e_commerce:done_urls "https://xxx.com/item/123"

3. 临时数据缓存（防丢失）

作用：电商数据采集后，若直接写入数据库时中断（如数据库宕机），会导致数据丢失。用 Redis 临时缓存已采集数据，待数据库恢复后批量入库。

数据结构选择：Hash，key 为 “任务 ID / 商品 ID”，value 为 JSON 格式的采集数据（便于后续解析）。

示例命令：

# 缓存已采集的商品数据（key=商品ID，value=JSON）
HSET e_commerce:temp_data "123" '{"title":"iPhone 15","price":5999,"stock":100}'
HSET e_commerce:temp_data "456" '{"title":"华为Mate 60","price":4999,"stock":50}'# 数据库恢复后，批量获取数据入库
HGETALL e_commerce:temp_data  # 返回所有临时数据
# 入库成功后删除临时数据
DEL e_commerce:temp_data

三、实战代码：Python + Redis 实现电商断点续爬

以 “采集某电商商品详情页” 为例，结合redis-py库实现完整断点续爬逻辑，框架采用通用爬虫结构（可适配 Scrapy、Requests 等）。

1. 环境准备

# 安装依赖
pip install redis requests beautifulsoup4

2. 完整代码实现

import redis
import requests
from bs4 import BeautifulSoup
import json# 1. 初始化Redis连接（开启持久化，避免断点数据丢失）
redis_client = redis.Redis(host="localhost",port=6379,db=0,decode_responses=True  # 自动解码为字符串，避免b''格式
)# 2. 配置Key前缀（避免与其他项目冲突）
WAIT_URLS_KEY = "e_commerce:wait_urls"  # 待爬URL队列
DONE_URLS_KEY = "e_commerce:done_urls"  # 已爬URL集合
TEMP_DATA_KEY = "e_commerce:temp_data"  # 临时数据缓存# 3. 初始化待爬队列（首次运行时添加初始URL，后续从Redis恢复）
def init_wait_urls(initial_urls):for url in initial_urls:# 先判断是否已在待爬/已爬队列，避免重复添加if not (redis_client.sismember(DONE_URLS_KEY, url) or url in redis_client.lrange(WAIT_URLS_KEY, 0, -1)):redis_client.lpush(WAIT_URLS_KEY, url)print(f"初始化完成，待爬URL数量：{redis_client.llen(WAIT_URLS_KEY)}")# 4. 采集函数（解析商品数据）
def crawl_item(url):try:response = requests.get(url, headers={"User-Agent": "Mozilla/5.0"})response.raise_for_status()  # 抛出HTTP错误soup = BeautifulSoup(response.text, "html.parser")# 解析商品数据（示例：标题、价格、商品ID）item_id = url.split("/")[-1]  # 从URL提取商品IDtitle = soup.find("h1", class_="product-title").get_text(strip=True)price = soup.find("span", class_="price").get_text(strip=True)return {"item_id": item_id,"title": title,"price": price,"url": url}except Exception as e:print(f"采集失败（{url}）：{str(e)}")# 采集失败的URL重新加入待爬队列（避免永久丢失）redis_client.lpush(WAIT_URLS_KEY, url)return None# 5. 主爬虫逻辑（断点续爬核心）
def run_crawler(initial_urls=None):# 首次运行初始化待爬队列，后续从Redis恢复if initial_urls and redis_client.llen(WAIT_URLS_KEY) == 0:init_wait_urls(initial_urls)print(f"开始爬虫，当前待爬URL数量：{redis_client.llen(WAIT_URLS_KEY)}")while True:# 1. 从Redis获取待爬URL（原子操作，避免多实例重复取）url = redis_client.rpop(WAIT_URLS_KEY)if not url:print("所有任务完成！")break# 2. 判断URL是否已爬（防止重复采集）if redis_client.sismember(DONE_URLS_KEY, url):print(f"URL已爬，跳过：{url}")continue# 3. 执行采集item_data = crawl_item(url)if not item_data:continue# 4. 缓存采集数据（待入库）redis_client.hset(TEMP_DATA_KEY, item_data["item_id"], json.dumps(item_data))print(f"采集成功，缓存数据：{item_data['title']}")# 5. 将URL加入已爬集合（标记完成）redis_client.sadd(DONE_URLS_KEY, url)# 6. （可选）批量入库：当临时数据达到100条时，写入数据库if redis_client.hlen(TEMP_DATA_KEY) >= 100:batch_insert_db()# 6. 批量入库函数（示例，需根据实际数据库适配）
def batch_insert_db():print("开始批量入库...")# 从Redis获取所有临时数据temp_data = redis_client.hgetall(TEMP_DATA_KEY)if not temp_data:return# 模拟数据库写入（实际项目替换为MySQL/MongoDB等）for item_id, data_json in temp_data.items():item = json.loads(data_json)# 执行入库SQL/API：如 insert into products (item_id, title, price) values (...)print(f"入库成功：{item['title']}")# 入库成功后，清空临时数据redis_client.delete(TEMP_DATA_KEY)print("批量入库完成！")# 7. 启动爬虫（首次运行需传入初始URL，后续无需）
if __name__ == "__main__":# 初始URL（如电商分类页下的商品列表URL）initial_urls = ["https://xxx.com/item/123","https://xxx.com/item/456","https://xxx.com/item/789"]# 启动爬虫（中断后重启，会自动从Redis恢复进度）run_crawler(initial_urls)

四、关键优化：适配电商场景的特殊需求

1. 分布式爬虫的任务分配

若需多台机器同时爬取（如采集百万级商品），可通过 Redis 实现分布式任务调度：

• 所有实例共享同一 Redis 的wait_urls队列；

• 利用 Redis 的原子命令RPOP（或SPOP），确保每个 URL 仅被一个实例获取；

• 已爬集合done_urls全局共享，避免多实例重复爬取。

2. 电商数据更新的处理

电商商品价格、库存会实时变化，仅靠 “已爬集合” 会导致旧数据无法更新。解决方案：

• 用Sorted Set存储已爬 URL，分数设为 “最后爬取时间戳”；

• 爬虫每次取 URL 前，先清理 “过期” 任务（如 7 天前爬过的 URL 重新加入待爬队列）：

import time
current_time = time.time()
# 找出7天前（604800秒）爬过的URL，重新加入待爬队列
expired_urls = redis_client.zrangebyscore(DONE_URLS_KEY, 0, current_time - 604800)
for url in expired_urls:redis_client.lpush(WAIT_URLS_KEY, url)redis_client.zrem(DONE_URLS_KEY, url)  # 从已爬集合删除

3. Redis 持久化配置（防数据丢失）

默认 Redis 数据存于内存，若 Redis 宕机，断点数据会丢失。需开启持久化：

1. AOF 持久化（推荐）：记录所有写命令，重启时重新执行命令恢复数据；

• • 配置文件redis.conf：appendonly yes，appendfsync everysec（每秒同步一次）；
    
    2.RDB 持久化：定时生成数据快照，适合备份；
• • 配置文件：save 3600 1（1 小时内至少 1 次写操作则生成快照）。

五、总结

利用 Redis 实现电商爬虫断点续爬，核心是通过 “待爬队列（List/Sorted Set）+ 已爬集合（Set）+ 临时缓存（Hash）” 三大模块，记录爬虫状态。相比传统文件存储（如 JSON/CSV），Redis 的高性能、原子操作、分布式支持，能更好应对电商场景的高并发、大数据量需求。

实际项目中，还需结合反爬策略（如 IP 池、UA 轮换）、数据库批量入库优化，进一步提升爬虫稳定性与效率。若需适配 Scrapy 框架，可直接使用scrapy-redis库（底层原理与本文一致，无需重复造轮子）。