当前位置：首页 > news >正文

如何动态调整Python爬虫的Request请求延迟

news 来源：原创 2025/4/22 13:44:54

引言

在网络爬虫开发中，合理控制请求延迟（Request Delay）是避免被封禁、提高爬取效率的关键。固定延迟（如 **time.sleep(1)**）虽然简单，但在面对不同网站的反爬策略时可能不够灵活。动态调整请求延迟能够更智能地适应目标网站的变化，提高爬虫的稳定性和效率。

本文将介绍如何动态调整Python爬虫的请求延迟，包括：

固定延迟 vs. 动态延迟的优劣
基于响应状态码的动态延迟调整
基于请求频率的动态延迟调整
结合代理IP和用户代理（User-Agent）优化延迟

1. 固定延迟 vs. 动态延迟

1.1 固定延迟

固定延迟是最简单的控制方式，例如：

import time
import requestsfor url in urls:response = requests.get(url)time.sleep(1)  # 固定延迟1秒

优点：实现简单，适用于低频率爬取。
缺点：

如果目标网站允许更快的请求，固定延迟会降低爬取效率。
如果目标网站检测到固定间隔请求，可能触发反爬机制。

1.2 动态延迟

动态延迟根据网站响应、请求频率等因素调整等待时间，例如：

如果服务器返回 **429 Too Many Requests**，则增加延迟。
如果连续多次请求成功，则适当降低延迟。
随机化延迟，模拟人类操作。

2. 基于响应状态码的动态延迟

如果服务器返回 **429** 或 **503**，说明请求频率过高，此时应增加延迟；如果正常返回 **200**，则可以适当降低延迟。

实现代码

import time
import requests
import randomclass DynamicDelayCrawler:def __init__(self, base_delay=1, max_delay=5):self.base_delay = base_delay  # 基础延迟self.max_delay = max_delay    # 最大延迟self.current_delay = base_delaydef adjust_delay(self, status_code):if status_code == 429:  # 请求过多，增加延迟self.current_delay = min(self.current_delay * 2, self.max_delay)elif status_code == 200:  # 请求成功，尝试降低延迟self.current_delay = max(self.current_delay * 0.9, self.base_delay)def crawl(self, url):try:response = requests.get(url)self.adjust_delay(response.status_code)print(f"URL: {url}, Status: {response.status_code}, Delay: {self.current_delay:.2f}s")time.sleep(self.current_delay)return response.textexcept Exception as e:print(f"Error fetching {url}: {e}")time.sleep(self.current_delay * 2)  # 出错时增加延迟return None# 测试
crawler = DynamicDelayCrawler(base_delay=1, max_delay=10)
urls = ["https://example.com/page1", "https://example.com/page2", "https://example.com/page3"]
for url in urls:crawler.crawl(url)

3. 基于请求频率的动态延迟

某些网站可能没有明确的 **429** 响应，但会通过其他方式限制爬虫（如封IP）。我们可以统计单位时间内的请求次数，动态调整延迟。

实现代码

import time
import requests
from collections import dequeclass RequestRateLimiter:def __init__(self, max_requests=10, time_window=10):self.max_requests = max_requests  # 时间窗口内允许的最大请求数self.time_window = time_window    # 时间窗口（秒）self.request_times = deque()      # 存储请求时间戳def wait_if_needed(self):now = time.time()# 移除超出时间窗口的请求记录while self.request_times and now - self.request_times[0] > self.time_window:self.request_times.popleft()if len(self.request_times) >= self.max_requests:# 计算需要等待的时间wait_time = self.time_window - (now - self.request_times[0])print(f"Rate limit reached, waiting {wait_time:.2f}s")time.sleep(wait_time)self.request_times.append(now)# 测试
limiter = RequestRateLimiter(max_requests=5, time_window=5)  # 5秒内最多5次请求
urls = [f"https://example.com/page{i}" for i in range(10)]
for url in urls:limiter.wait_if_needed()response = requests.get(url)print(f"Fetched {url}, Status: {response.status_code}")

4. 结合代理IP和随机User-Agent优化

动态调整延迟的同时，使用代理IP和随机User-Agent可以进一步降低被封禁的风险。

实现代码

import random
import time
import requests
from fake_useragent import UserAgentclass AdvancedCrawler:def __init__(self, base_delay=1, max_delay=10):self.base_delay = base_delayself.max_delay = max_delayself.current_delay = base_delayself.ua = UserAgent()# 添加指定的代理信息self.proxyHost = "www.16yun.cn"self.proxyPort = "5445"self.proxyUser = "16QMSOML"self.proxyPass = "280651"self.proxies = [f"http://{self.proxyUser}:{self.proxyPass}@{self.proxyHost}:{self.proxyPort}",# 如果需要保留原有代理，可以将它们也加入到列表中# "<url id="d02v8neruqkqvdqddo90" type="url" status="failed" title="" wc="0">http://proxy1.example.com:8080</url> ",# "<url id="d02v8neruqkqvdqddo9g" type="url" status="failed" title="" wc="0">http://proxy2.example.com:8080</url> ",]def get_random_proxy(self):return random.choice(self.proxies) if self.proxies else Nonedef adjust_delay(self, status_code):if status_code == 429:self.current_delay = min(self.current_delay * 2, self.max_delay)elif status_code == 200:self.current_delay = max(self.current_delay * 0.9, self.base_delay)def crawl(self, url):headers = {"User-Agent": self.ua.random}proxy = self.get_random_proxy()try:response = requests.get(url,headers=headers,proxies={"http": proxy, "https": proxy} if proxy else None,timeout=10)self.adjust_delay(response.status_code)print(f"URL: {url}, Status: {response.status_code}, Delay: {self.current_delay:.2f}s")time.sleep(self.current_delay + random.uniform(0, 0.5))  # 增加随机抖动return response.textexcept Exception as e:print(f"Error fetching {url}: {e}")time.sleep(self.current_delay * 2)return None# 测试
crawler = AdvancedCrawler(base_delay=1, max_delay=10)
urls = [f"https://example.com/page{i}" for i in range(5)]
for url in urls:crawler.crawl(url)

5总结

动态调整Python爬虫的Request请求延迟是一种有效的优化策略，可以提高爬虫的稳定性和效率。通过基于响应时间、服务器负载和反爬机制的动态调整策略，爬虫可以在复杂的网络环境中灵活运行，同时降低被封禁的风险。本文提供的代码示例展示了如何实现动态调整请求延迟，开发者可以根据实际需求进行进一步优化和扩展。