引言
在网络爬虫开发中，合理控制请求延迟（Request Delay）是避免被封禁、提高爬取效率的关键。固定延迟（如 time.sleep(1)）虽然简单，但在面对不同网站的反爬策略时可能不够灵活。动态调整请求延迟能够更智能地适应目标网站的变化，提高爬虫的稳定性和效率。
本文将介绍如何动态调整Python爬虫的请求延迟，包括：
- 固定延迟 vs. 动态延迟的优劣
- 基于响应状态码的动态延迟调整
- 基于请求频率的动态延迟调整
- 结合代理IP和用户代理（User-Agent）优化延迟
- 固定延迟 vs. 动态延迟
1.1 固定延迟
固定延迟是最简单的控制方式，例如：
import time
import requests
for url in urls:
response = requests.get(url)
time.sleep(1) # 固定延迟1秒
优点：实现简单，适用于低频率爬取。
缺点：
● 如果目标网站允许更快的请求，固定延迟会降低爬取效率。
● 如果目标网站检测到固定间隔请求，可能触发反爬机制。
1.2 动态延迟
动态延迟根据网站响应、请求频率等因素调整等待时间，例如：
● 如果服务器返回 429 Too Many Requests，则增加延迟。
● 如果连续多次请求成功，则适当降低延迟。
● 随机化延迟，模拟人类操作。
- 基于响应状态码的动态延迟
如果服务器返回 429 或 503，说明请求频率过高，此时应增加延迟；如果正常返回 200，则可以适当降低延迟。
实现代码
import time
import requests
import random
class DynamicDelayCrawler:
def init(self, base_delay=1, max_delay=5):
self.base_delay = base_delay # 基础延迟
self.max_delay = max_delay # 最大延迟
self.current_delay = base_delay
def 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.text
except 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)
- 基于请求频率的动态延迟
某些网站可能没有明确的 429 响应，但会通过其他方式限制爬虫（如封IP）。我们可以统计单位时间内的请求次数，动态调整延迟。
实现代码
import time
import requests
from collections import deque
class 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}")
- 结合代理IP和随机User-Agent优化
动态调整延迟的同时，使用代理IP和随机User-Agent可以进一步降低被封禁的风险。
实现代码
import random
import time
import requests
from fake_useragent import UserAgent
class AdvancedCrawler:
def init(self, base_delay=1, max_delay=10):
self.base_delay = base_delay
self.max_delay = max_delay
self.current_delay = base_delay
self.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 None
def 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.text
except 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请求延迟是一种有效的优化策略，可以提高爬虫的稳定性和效率。通过基于响应时间、服务器负载和反爬机制的动态调整策略，爬虫可以在复杂的网络环境中灵活运行，同时降低被封禁的风险。本文提供的代码示例展示了如何实现动态调整请求延迟，开发者可以根据实际需求进行进一步优化和扩展。
