Python 爬虫实战：爬虫请求超时设置与重试机制

前言

在爬虫开发中，网络环境的不稳定性（如延迟、断连）、目标服务器的限流策略、反爬机制触发等问题，极易导致请求超时或失败。若缺乏有效的超时控制和重试机制，爬虫程序可能陷入无限等待、频繁崩溃，甚至被目标网站封禁 IP。超时设置是避免请求阻塞的 “安全阀”，重试机制则是提升爬虫容错性的 “缓冲垫”，二者结合是保障爬虫稳定运行的核心手段。本文将从超时原理、重试策略入手，以「京东商品列表页」（https://list.jd.com/list.html?cat=652,12345,12346）为实战目标，系统讲解请求超时的精细化设置、多场景重试机制的实现，以及超时与重试的协同优化方案，帮助开发者打造高可用的爬虫程序。

摘要

本文以京东商品数据爬取为实战场景，深入剖析 Python 爬虫请求超时的类型（连接超时、读取超时）及设置原理，详细讲解固定次数重试、指数退避重试、条件触发重试等核心重试策略的实现方式。通过实战代码演示如何结合requests库的超时参数与自定义重试逻辑，解决网络波动、服务器响应慢、临时反爬等问题，并给出超时阈值调优、重试次数把控、失败降级处理等进阶方案，最终实现 “超时可控、重试有度、失败有兜底” 的稳健爬虫架构。

一、请求超时与重试机制核心认知

1.1 请求超时的类型与原理

爬虫发送 HTTP 请求时，超时可分为两个阶段，核心差异如下表所示：

核心原理：requests库的timeout参数支持两种设置方式：

• 单值设置（如timeout=10）：同时控制连接超时和读取超时；
• 双值设置（如timeout=(3, 7)）：分别设置连接超时（3 秒）和读取超时（7 秒），精细化控制不同阶段的等待时间。

1.2 重试机制的核心价值与设计原则

（1）核心价值

• 应对临时网络波动：如短时间断网、服务器瞬时过载；
• 规避轻度反爬策略：如目标网站临时限流导致的请求失败；
• 提升数据采集完整性：减少因单次失败导致的数据丢失。

（2）设计原则

1.3 常见可重试 / 不可重试异常分类

二、实战代码实现：超时设置与重试机制

2.1 完整代码

python

运行

import requests from bs4 import BeautifulSoup import csv import time import logging from fake_useragent import UserAgent from typing import Optional, List, Dict, Tuple from requests.exceptions import RequestException, ConnectTimeout, ReadTimeout, HTTPError # ====================== 日志配置 ====================== logging.basicConfig( level=logging.INFO, format="%(asctime)s - %(levelname)s - %(filename)s:%(lineno)d - %(message)s", handlers=[ logging.StreamHandler(), # 控制台输出 logging.FileHandler("jd_crawler.log", encoding="utf-8") # 日志文件 ] ) logger = logging.getLogger(__name__) # ====================== 核心配置 ====================== # 目标URL（京东笔记本电脑列表页） BASE_URL = "https://list.jd.com/list.html?cat=670,671,672&page={}" # 爬取页码范围 PAGE_RANGE = (1, 10) # CSV存储路径 CSV_PATH = "jd_laptop.csv" # 请求超时配置（连接超时3秒，读取超时7秒） TIMEOUT_CONFIG: Tuple[int, int] = (3, 7) # 重试机制配置 RETRY_CONFIG = { "max_retry": 3, # 最大重试次数 "base_delay": 2, # 基础休眠时间（秒） "backoff_factor": 1.5, # 指数退避因子（每次休眠时间 = base_delay * (backoff_factor ^ 重试次数)） "retry_exceptions": (ConnectTimeout, ReadTimeout, ConnectionError) # 可重试异常 } # 请求头配置 HEADERS = { "User-Agent": UserAgent().random, "Accept-Language": "zh-CN,zh;q=0.9", "Accept-Encoding": "gzip, deflate, br", "Referer": "https://www.jd.com/", "Connection": "keep-alive" } # ====================== 重试装饰器实现 ====================== def retry_decorator(max_retry: int, base_delay: float, backoff_factor: float, retry_exceptions: tuple): """ 通用重试装饰器：支持指数退避休眠、指定可重试异常 :param max_retry: 最大重试次数 :param base_delay: 基础休眠时间 :param backoff_factor: 指数退避因子 :param retry_exceptions: 可重试异常元组 """ def decorator(func): def wrapper(*args, **kwargs): retry_count = 0 while retry_count < max_retry: try: # 执行目标函数 return func(*args, **kwargs) except retry_exceptions as e: # 捕获可重试异常，执行重试逻辑 retry_count += 1 # 计算指数退避休眠时间 sleep_time = base_delay * (backoff_factor ** (retry_count - 1)) logger.warning( f"【重试触发】函数{func.__name__}执行失败（第{retry_count}/{max_retry}次），" f"异常类型：{type(e).__name__}，异常信息：{str(e)[:80]}，" f"休眠{sleep_time:.2f}秒后重试" ) time.sleep(sleep_time) except HTTPError as e: # 处理HTTP状态码异常 status_code = e.response.status_code if hasattr(e, 'response') else None if status_code and 500 <= status_code < 600: # 5xx状态码：执行重试 retry_count += 1 sleep_time = base_delay * (backoff_factor ** (retry_count - 1)) logger.warning( f"【重试触发】HTTP状态码异常{status_code}（第{retry_count}/{max_retry}次），" f"休眠{sleep_time:.2f}秒后重试" ) time.sleep(sleep_time) else: # 4xx状态码：直接抛出，不重试 logger.error(f"【不可重试】HTTP状态码异常{status_code}，终止重试") raise except Exception as e: # 其他不可重试异常：直接抛出 logger.error(f"【不可重试】未知异常{type(e).__name__}，终止重试，异常信息：{str(e)[:80]}") raise # 重试耗尽仍失败 logger.error(f"【重试耗尽】函数{func.__name__}执行失败，已重试{max_retry}次，终止执行") return None return wrapper return decorator # ====================== 爬虫核心函数 ====================== @retry_decorator(**RETRY_CONFIG) def fetch_jd_page(page_num: int) -> Optional[str]: """ 获取京东商品列表页HTML（带超时设置和重试机制） :param page_num: 页码 :return: 页面HTML文本，失败返回None """ url = BASE_URL.format(page_num) logger.info(f"【请求页面】页码{page_num}，URL：{url}") # 发送请求：设置精细化超时参数 response = requests.get( url=url, headers=HEADERS, timeout=TIMEOUT_CONFIG, # (连接超时3秒，读取超时7秒) allow_redirects=True, # 允许重定向 verify=False # 忽略SSL证书验证（京东部分页面需此配置） ) # 校验响应状态码（非2xx会抛出HTTPError） response.raise_for_status() # 处理编码：京东页面多为gbk编码 response.encoding = "gbk" if response.encoding == "ISO-8859-1" else response.encoding logger.info(f"【请求成功】页码{page_num}，响应状态码：{response.status_code}") return response.text def parse_jd_page(html: str) -> List[Dict]: """ 解析京东商品列表页数据 :param html: 页面HTML文本 :return: 商品数据列表，解析失败返回空列表 """ if not html: logger.warning("【解析失败】HTML文本为空") return [] product_list = [] try: soup = BeautifulSoup(html, "html.parser") # 定位商品条目 product_items = soup.find_all("li", class_="gl-item") for item in product_items: # 提取核心字段（容错处理） product_id = item.get("data-sku", "未知") # 商品名称 name_tag = item.find("div", class_="p-name").find("em") product_name = name_tag.text.strip().replace("\n", "").replace("\t", "") if name_tag else "未知" # 商品价格 price_tag = item.find("div", class_="p-price").find("i") product_price = price_tag.text.strip() if price_tag else "0.00" # 店铺名称 shop_tag = item.find("div", class_="p-shop").find("a") shop_name = shop_tag.text.strip() if shop_tag else "未知" # 构造商品数据字典 product_info = { "页码": page_num, "商品ID": product_id, "商品名称": product_name, "价格(元)": product_price, "店铺名称": shop_name } product_list.append(product_info) logger.info(f"【解析成功】页码{page_num}，提取{len(product_list)}条商品数据") except Exception as e: logger.error(f"【解析失败】页码{page_num}，异常：{e}", exc_info=True) return product_list def batch_save_to_csv(data: List[Dict], file_path: str, is_append: bool = True): """ 批量保存商品数据到CSV :param data: 商品数据列表 :param file_path: CSV路径 :param is_append: 是否追加模式 """ if not data: logger.warning("【存储失败】无商品数据可写入") return # 定义CSV表头 headers = ["页码", "商品ID", "商品名称", "价格(元)", "店铺名称"] # 选择文件打开模式 mode = "a+" if is_append else "w" try: with open(file_path, mode, newline="", encoding="utf-8-sig") as f: writer = csv.DictWriter(f, fieldnames=headers) # 首次写入时添加表头 if f.tell() == 0: writer.writeheader() # 批量写入数据 writer.writerows(data) logger.info(f"【存储成功】共写入{len(data)}条商品数据到{file_path}") except Exception as e: logger.error(f"【存储失败】写入CSV异常：{e}") def crawl_jd_laptop(): """ 爬取京东笔记本电脑列表页主函数 """ logger.info("【爬虫启动】开始爬取京东笔记本电脑数据") total_success = 0 # 累计成功爬取的商品数 for page_num in range(PAGE_RANGE[0], PAGE_RANGE[1] + 1): # 1. 获取页面HTML html = fetch_jd_page(page_num) if not html: logger.warning(f"【页面获取失败】页码{page_num}，跳过该页") continue # 2. 解析页面数据 product_data = parse_jd_page(html) if not product_data: logger.warning(f"【数据解析失败】页码{page_num}，跳过该页") continue # 3. 批量存储数据 batch_save_to_csv(product_data, CSV_PATH) total_success += len(product_data) # 4. 控制爬取频率（避免反爬） time.sleep(1.5) logger.info(f"【爬虫结束】爬取完成，累计获取{total_success}条商品数据") # ====================== 程序入口 ====================== if __name__ == "__main__": try: # 忽略SSL警告（因verify=False产生） requests.packages.urllib3.disable_warnings(requests.packages.urllib3.exceptions.InsecureRequestWarning) # 执行爬虫主函数 crawl_jd_laptop() except KeyboardInterrupt: logger.info("【爬虫终止】用户手动终止程序") except Exception as e: logger.critical(f"【爬虫崩溃】全局未捕获异常：{e}", exc_info=True)

2.2 核心逻辑解析

（1）精细化超时设置

• 参数设计：采用(3, 7)双值超时配置，连接阶段最多等待 3 秒（TCP 连接建立是基础，超时过久无意义），读取阶段最多等待 7 秒（服务器返回数据可能较慢，需预留合理时间）；
• 场景适配：针对京东服务器响应特性，设置verify=False忽略 SSL 证书验证，避免因证书问题导致的连接超时；
• 编码处理：京东页面默认编码为 gbk，通过判断response.encoding == "ISO-8859-1"自动修正编码，避免乱码导致的后续解析超时。

（2）指数退避重试机制

• 核心实现：重试装饰器中通过base_delay * (backoff_factor ^ 重试次数)计算休眠时间，例如：
  • 第 1 次重试：2 * (1.5^0) = 2 秒；
  • 第 2 次重试：2 * (1.5^1) = 3 秒；
  • 第 3 次重试：2 * (1.5^2) = 4.5 秒；
• 优势：相比固定间隔重试，指数退避能降低短时间内高频请求的风险，更适配服务器限流策略；
• 异常精准控制：仅对ConnectTimeout、ReadTimeout、ConnectionError及 5xx 状态码执行重试，4xx 异常直接终止，避免无效重试。

（3）重试装饰器的通用性

• 参数解耦：通过**RETRY_CONFIG将重试配置与装饰器解耦，便于后续调整重试次数、休眠时间等参数；
• 日志追溯：记录每次重试的次数、休眠时间、异常类型，问题发生时可快速定位根因；
• 代码复用：装饰器可直接复用至其他爬虫函数，无需重复编写重试逻辑。

2.3 代码运行结果

（1）控制台输出示例

plaintext

2025-12-17 14:00:00,123 - INFO - jd_crawler.py:89 - 【爬虫启动】开始爬取京东笔记本电脑数据 2025-12-17 14:00:00,124 - INFO - jd_crawler.py:65 - 【请求页面】页码1，URL：https://list.jd.com/list.html?cat=670,671,672&page=1 2025-12-17 14:00:01,234 - INFO - jd_crawler.py:78 - 【请求成功】页码1，响应状态码：200 2025-12-17 14:00:01,345 - INFO - jd_crawler.py:95 - 【解析成功】页码1，提取30条商品数据 2025-12-17 14:00:01,456 - INFO - jd_crawler.py:125 - 【存储成功】共写入30条商品数据到jd_laptop.csv 2025-12-17 14:00:03,567 - INFO - jd_crawler.py:65 - 【请求页面】页码2，URL：https://list.jd.com/list.html?cat=670,671,672&page=2 2025-12-17 14:00:06,678 - WARNING - jd_crawler.py:45 - 【重试触发】函数fetch_jd_page执行失败（第1/3次），异常类型：ReadTimeout，异常信息：HTTPSConnectionPool(host='list.jd.com', port=443): Read timed out. (read timeout=7)，休眠2.00秒后重试 2025-12-17 14:00:09,789 - INFO - jd_crawler.py:78 - 【请求成功】页码2，响应状态码：200 2025-12-17 14:00:09,890 - INFO - jd_crawler.py:95 - 【解析成功】页码2，提取30条商品数据 2025-12-17 14:00:09,901 - INFO - jd_crawler.py:125 - 【存储成功】共写入30条商品数据到jd_laptop.csv ... 2025-12-17 14:02:30,123 - INFO - jd_crawler.py:145 - 【爬虫结束】爬取完成，累计获取285条商品数据

（2）日志文件（jd_crawler.log）关键内容

plaintext

2025-12-17 14:00:06,678 - WARNING - jd_crawler.py:45 - 【重试触发】函数fetch_jd_page执行失败（第1/3次），异常类型：ReadTimeout，异常信息：HTTPSConnectionPool(host='list.jd.com', port=443): Read timed out. (read timeout=7)，休眠2.00秒后重试 2025-12-17 14:00:09,789 - INFO - jd_crawler.py:78 - 【请求成功】页码2，响应状态码：200 2025-12-17 14:00:15,123 - ERROR - jd_crawler.py:58 - 【重试耗尽】函数fetch_jd_page执行失败，已重试3次，终止执行 2025-12-17 14:00:15,234 - WARNING - jd_crawler.py:139 - 【页面获取失败】页码5，跳过该页

（3）CSV 文件输出示例

三、超时与重试机制进阶优化

3.1 动态超时阈值调整

根据请求成功率动态调整超时阈值，适配不同网络环境：

python

运行

def dynamic_timeout_adjust(success_rate: float) -> Tuple[int, int]: """ 动态调整超时阈值 :param success_rate: 近期请求成功率（0-1） :return: 新的超时配置（连接超时，读取超时） """ if success_rate > 0.9: # 成功率高，适当降低超时阈值，提升效率 return (2, 5) elif 0.7 <= success_rate <= 0.9: # 成功率中等，保持默认阈值 return (3, 7) else: # 成功率低，提高超时阈值，降低失败率 return (5, 10) # 调用示例：统计近期请求成功率并调整 success_count = 0 total_count = 0 def update_success_rate(success: bool): global success_count, total_count total_count += 1 if success: success_count += 1 success_rate = success_count / total_count if total_count > 0 else 1.0 global TIMEOUT_CONFIG TIMEOUT_CONFIG = dynamic_timeout_adjust(success_rate) logger.info(f"【超时调整】当前成功率{success_rate:.2f}，新超时配置：{TIMEOUT_CONFIG}") # 在fetch_jd_page函数中调用 if html: update_success_rate(True) else: update_success_rate(False)

3.2 重试机制与代理 IP 协同

重试时自动切换代理 IP，提升重试成功率：

python

运行

# 代理IP池（示例，实际需从代理服务商获取） PROXY_POOL = [ {"http": "http://127.0.0.1:8080", "https": "https://127.0.0.1:8080"}, {"http": "http://127.0.0.1:8081", "https": "https://127.0.0.1:8081"}, {"http": "http://127.0.0.1:8082", "https": "https://127.0.0.1:8082"} ] current_proxy_idx = 0 def get_next_proxy() -> dict: """获取下一个代理IP""" global current_proxy_idx proxy = PROXY_POOL[current_proxy_idx] current_proxy_idx = (current_proxy_idx + 1) % len(PROXY_POOL) return proxy # 修改fetch_jd_page函数，重试时切换代理 @retry_decorator(**RETRY_CONFIG) def fetch_jd_page(page_num: int) -> Optional[str]: url = BASE_URL.format(page_num) logger.info(f"【请求页面】页码{page_num}，URL：{url}") # 获取当前代理 proxy = get_next_proxy() if retry_count > 0 else {} response = requests.get( url=url, headers=HEADERS, timeout=TIMEOUT_CONFIG, proxies=proxy, # 添加代理 allow_redirects=True, verify=False ) # ... 其余逻辑不变

3.3 超时 / 重试监控与告警

当超时 / 重试频率超过阈值时，发送告警通知：

python

运行

import smtplib from email.mime.text import MIMEText from email.header import Header # 告警配置 ALERT_THRESHOLD = { "timeout_rate": 0.3, # 超时率阈值（超过30%告警） "retry_rate": 0.5 # 重试率阈值（超过50%告警） } ALERT_EMAIL = { "sender": "your_email@163.com", "password": "your_email_auth_code", "receiver": "admin@xxx.com", "smtp_server": "smtp.163.com" } def send_alert(alert_type: str, rate: float): """发送超时/重试告警邮件""" subject = f"【爬虫告警】{alert_type}率超标，当前值：{rate:.2f}" content = f""" 爬虫告警通知： 告警类型：{alert_type}率超标 当前值：{rate:.2f} 阈值：{ALERT_THRESHOLD[f"{alert_type}_rate"]} 时间：{time.strftime('%Y-%m-%d %H:%M:%S')} """ message = MIMEText(content, "plain", "utf-8") message["From"] = Header("京东爬虫监控", "utf-8") message["To"] = Header("运维人员", "utf-8") message["Subject"] = Header(subject, "utf-8") try: smtp_obj = smtplib.SMTP_SSL(ALERT_EMAIL["smtp_server"], 465) smtp_obj.login(ALERT_EMAIL["sender"], ALERT_EMAIL["password"]) smtp_obj.sendmail(ALERT_EMAIL["sender"], ALERT_EMAIL["receiver"], message.as_string()) smtp_obj.quit() logger.info(f"【告警发送成功】{alert_type}率超标告警已发送") except Exception as e: logger.error(f"【告警发送失败】异常：{e}") # 监控函数 timeout_count = 0 retry_count_total = 0 total_request_count = 0 def monitor_metrics(is_timeout: bool, is_retry: bool): """监控超时率和重试率""" global timeout_count, retry_count_total, total_request_count total_request_count += 1 if is_timeout: timeout_count += 1 if is_retry: retry_count_total += 1 # 计算比率 timeout_rate = timeout_count / total_request_count if total_request_count > 0 else 0.0 retry_rate = retry_count_total / total_request_count if total_request_count > 0 else 0.0 # 触发告警 if timeout_rate > ALERT_THRESHOLD["timeout_rate"]: send_alert("timeout", timeout_rate) if retry_rate > ALERT_THRESHOLD["retry_rate"]: send_alert("retry", retry_rate)

3.4 超时与重试机制性能对比

四、常见问题与解决方案

4.1 重试机制导致的重复爬取

• 问题现象：重试成功后，同一页面数据被重复写入 CSV；
• 解决方案：
  1. 为每条数据添加唯一标识（如商品 ID），存储前先校验是否已存在；
  2. 记录已成功爬取的页码，重试时跳过已爬取成功的页码；
  3. 使用数据库（如 SQLite）存储数据，通过唯一键约束避免重复。

4.2 超时设置过短 / 过长的问题

4.3 指数退避重试导致的爬取延迟

• 问题现象：多次重试后，休眠时间过长，爬取进度缓慢；
• 解决方案：
  1. 设置休眠时间上限（如最大休眠 5 秒），避免指数增长导致的过度延迟；
  2. 对不同异常设置不同退避因子（如超时异常退避因子 1.5，连接错误退避因子 2.0）；
  3. 批量爬取时，将重试任务放入独立队列，不阻塞主爬取流程。

五、总结与最佳实践

5.1 核心总结

本文通过京东商品爬取实战，完整实现了 “精细化超时设置 + 指数退避重试” 的核心机制，关键要点包括：

1. 超时设置精细化：区分连接超时和读取超时，根据目标网站特性调整阈值；
2. 重试策略差异化：仅对可重试异常执行重试，采用指数退避休眠降低服务器压力；
3. 机制解耦复用：通过装饰器封装重试逻辑，提升代码复用性和可维护性；
4. 监控与优化：结合动态调整、代理切换、告警机制，提升超时 / 重试策略的适配性。

5.2 最佳实践清单

5.3 行业应用建议

1. 小规模爬取：使用固定超时 + 固定重试，兼顾效率与容错；
2. 中大规模爬取：采用动态超时 + 指数退避 + 代理切换，保障稳定性；
3. 生产环境爬取：增加监控告警、断点续爬、数据去重，实现全链路管控；
4. 高反爬网站爬取：结合 Cookie 池、User-Agent 池、分布式爬取，降低超时 / 重试频率。

超时设置和重试机制是爬虫稳定性的 “双保险”，但需注意：二者并非 “越多越好”—— 超时过久会降低效率，重试过多会加剧服务器压力。开发者需根据目标网站的特性、网络环境、业务需求，平衡 “容错性” 与 “效率”，才能设计出既稳定又高效的爬虫程序。