介绍
网络爬虫（Web Crawler）是自动化的数据采集工具，用于从网络上提取所需的数据。然而，随着反爬虫技术的不断进步，很多网站增加了复杂的防护机制，使得数据采集变得更加困难。在这种情况下，Python 的 requests 库因其易用性和强大的功能，成为了开发爬虫的常用工具。然而，在复杂的 HTTP 请求场景中，标准的 requests 使用往往不够灵活，爬虫需要结合代理、会话控制、限流等高级技巧来更好地适应不同网站的反爬限制。
本文将针对三种典型的复杂 HTTP 请求场景，分别为 Spider Trap（蜘蛛陷阱）、SESSION访问限制和请求频率限制，进行深入的技术分析，并给出实际代码示例，帮助读者掌握 Python Requests 的高级用法。
技术分析
1. 应对 Spider Trap（蜘蛛陷阱）
Spider Trap 是一种通过设置大量链接或无限循环的链接结构来消耗爬虫资源的技术。爬虫陷入这些陷阱后，可能会在特定网页中无限循环，导致资源浪费，甚至引发封禁。为避免此问题，我们可以采取以下措施：
- CSS类链接数控制：限制同一页面中每个 CSS 类中能爬取的最大链接数，从而防止在陷阱页面中过度抓取。
- URL去重：通过哈希或布隆过滤器（Bloom Filter）对已访问的 URL 进行去重，避免重复抓取。
以下代码展示了如何通过 Python Requests 结合代理和 CSS 类链接数控制来实现对 Spider Trap 的防护。
import requests
from bs4 import BeautifulSoup
import hashlib
# 代理信息，需替换成实际的亿牛云爬虫代理配置 www.16yun.cn
proxy = {
"http": "http://username:password@proxy.16yun.cn:18000",
"https": "http://username:password@proxy.16yun.cn:18000"
}
# 用于记录访问过的URL
visited_urls = set()
# 爬取函数
def crawl(url, max_links_per_class=10):
if url in visited_urls:
print(f"已访问过 URL：{url}")
return
try:
# 使用代理发送请求
response = requests.get(url, proxies=proxy, timeout=10)
response.raise_for_status()
visited_urls.add(url) # 标记该 URL 为已访问
soup = BeautifulSoup(response.text, 'html.parser')
# 统计每个 CSS 类中的链接数
class_link_count = {
}
for link in soup.find_all("a", href=True):
# 计算每个链接的哈希值
link_url = link['href']
link_class = link.get("class", [""])[0]
# 更新每个 CSS 类的链接数
class_link_count[link_class] = class_link_count.get(link_class, 0) + 1
# 超过最大链接数则跳过
if class_link_count[link_class] > max_links_per_class:
print(f"跳过过多链接的类：{link_class}")
continue
# 递归爬取新链接
if link_url not in visited_urls:
crawl(link_url)
except requests.RequestException as e:
print(f"请求失败：{e}")
# 开始爬取
crawl("http://example.com")
AI 代码解读
2. SESSION访问限制
某些网站会通过观察用户的操作模式来区分是普通用户还是爬虫。例如，频繁的请求和重复性高的操作可能被视为异常，导致账号被限制访问。针对这种情况，我们可以模拟用户的正常操作，比如在页面之间设置合理的等待时间，同时通过多个账号轮换访问来减少单一 SESSION 的负载。
以下代码示例展示了如何模拟多账号登录，并进行合理的延时，避免触发访问限制。
import requests
import time
import random
# 代理信息，需替换成实际的亿牛云爬虫代理配置 www.16yun.cn
proxy = {
"http": "http://username:password@proxy.16yun.cn:18000",
"https": "http://username:password@proxy.16yun.cn:18000"
}
# 用户账号列表
accounts = [
{
"username": "user1", "password": "pass1"},
{
"username": "user2", "password": "pass2"},
# 可添加更多账号
]
def login(account):
session = requests.Session()
login_url = "http://example.com/login"
try:
# 使用POST方法模拟登录请求
response = session.post(login_url, data=account, proxies=proxy)
response.raise_for_status()
# 检查登录状态
if "欢迎" in response.text:
print(f"{account['username']} 登录成功")
return session
else:
print(f"{account['username']} 登录失败")
return None
except requests.RequestException as e:
print(f"请求失败：{e}")
return None
# 主循环
for account in accounts:
session = login(account)
if session:
# 模拟正常操作，随机延时
for _ in range(5):
try:
response = session.get("http://example.com/data", proxies=proxy)
response.raise_for_status()
print(f"获取数据：{response.text[:100]}...")
# 模拟用户的延时
time.sleep(random.uniform(1, 3))
except requests.RequestException as e:
print(f"请求失败：{e}")
# 切换账号
time.sleep(random.uniform(5, 10))
AI 代码解读
3. 请求频率限制
为了防止频繁请求造成服务器压力，很多网站设置了请求频率限制。常用的限流算法包括令牌桶和漏桶。这些算法通过控制请求速度和时间间隔来实现稳定的数据请求。
以下示例展示了通过限流控制请求频率的方式，以避免触发请求频率限制。
import requests
import time
import threading
import queue
# 代理信息，需替换成实际的亿牛云爬虫代理配置 www.16yun.cn
proxy = {
"http": "http://username:password@proxy.16yun.cn:18000",
"https": "http://username:password@proxy.16yun.cn:18000"
}
# 配置令牌桶
class TokenBucket:
def __init__(self, rate, capacity):
self.rate = rate # 每秒生成的令牌数
self.capacity = capacity # 桶的最大容量
self.tokens = capacity
self.last_refill_time = time.time()
def acquire(self):
current_time = time.time()
elapsed = current_time - self.last_refill_time
# 更新令牌数量
self.tokens = min(self.capacity, self.tokens + elapsed * self.rate)
self.last_refill_time = current_time
# 判断是否可以进行请求
if self.tokens >= 1:
self.tokens -= 1
return True
else:
return False
# 创建令牌桶，限制每秒最多3个请求
token_bucket = TokenBucket(rate=3, capacity=3)
def limited_request(url):
while not token_bucket.acquire():
time.sleep(0.1) # 等待令牌
try:
response = requests.get(url, proxies=proxy)
print(f"获取数据：{response.text[:100]}...")
except requests.RequestException as e:
print(f"请求失败：{e}")
# 测试请求
urls = ["http://example.com/data"] * 10
for url in urls:
threading.Thread(target=limited_request, args=(url,)).start()
AI 代码解读
结论
本文深入探讨了 Python Requests 的高级使用技巧，帮助读者在面对复杂的 HTTP 请求场景时更加得心应手。通过代理的使用、CSS 类链接数控制、多账号 SESSION 切换、以及限流算法的实现，我们可以大幅提高爬虫的稳定性与效率，提升在反爬虫环境中的生存能力。掌握这些技巧不仅有助于提高抓取数据的成功率，同时也为更复杂的反爬需求打下了扎实的技术基础。
