1. 为什么爬虫需要限速与礼貌访问刚入门的爬虫开发者最容易犯的错误就是不考虑目标网站的承受能力直接开启暴力抓取模式。我见过太多新手写的爬虫把小型网站直接打挂的案例——这不仅是技术问题更是职业道德问题。合理的限速策略应该像老司机开车一样知道什么时候该加速什么时候必须刹车。服务器对异常流量的识别非常敏感。当你的请求频率超过正常人类操作的速度时比如1秒内发起几十次请求服务器会立即标记你的IP为爬虫。轻则返回429 Too Many Requests错误重则直接封禁IP。更严重的是这种粗暴的访问会挤占正常用户的带宽资源这就是为什么网络热词中会出现把正规爬虫挤得都没带宽了这样的吐槽。2. 基础限速方案实现2.1 固定延迟控制最简单的限速方法是在每个请求之间插入固定时间间隔。requests库本身没有内置限速功能但配合time模块就能实现import time import requests def crawl_with_delay(urls, delay1): for url in urls: response requests.get(url) print(f抓取 {url} 状态码: {response.status_code}) time.sleep(delay) # 关键延迟控制这个delay参数就是两次请求之间的间隔秒数。根据我的实测经验对于大多数资讯类网站设置为1-2秒是比较安全的阈值。但要注意几个关键点重要提示不要使用random.random()这类纯随机延迟因为真正的用户行为延迟是有规律可循的完全随机反而会被识别为机器人行为。应该使用类似random.uniform(0.8, 1.2)这样的范围随机。2.2 动态延迟调整更高级的做法是根据服务器响应状态动态调整延迟。当出现429状态码时自动增加延迟时间def adaptive_delay_crawl(urls): base_delay 1 for url in urls: try: response requests.get(url) if response.status_code 429: base_delay * 2 # 指数退避 print(f触发限流延迟增加到 {base_delay}秒) else: base_delay max(1, base_delay * 0.9) # 缓慢恢复 time.sleep(base_delay) except Exception as e: print(f请求异常: {str(e)}) time.sleep(60) # 出现异常时长时间等待这种算法被称为指数退避是处理限流的经典策略。当遇到429错误时延迟时间会成倍增加直到请求成功后再逐步恢复。3. 并发控制技术3.1 线程池限流使用concurrent.futures实现带并发数限制的爬取from concurrent.futures import ThreadPoolExecutor import queue def threaded_crawl(urls, max_workers3): url_queue queue.Queue() for url in urls: url_queue.put(url) def worker(): while not url_queue.empty(): url url_queue.get() response requests.get(url) print(f{url} → {response.status_code}) time.sleep(1) # 保持基础延迟 with ThreadPoolExecutor(max_workersmax_workers) as executor: for _ in range(max_workers): executor.submit(worker)这里max_workers控制着最大并发线程数。根据服务器性能不同一般建议设置在3-5之间。我在电商网站抓取实践中发现超过5个并发线程就很容易触发反爬机制。3.2 令牌桶算法实现更精确的流量控制可以使用令牌桶算法。下面是一个简化实现import threading class TokenBucket: def __init__(self, rate, capacity): self.rate rate # 令牌产生速率(个/秒) self.capacity capacity # 桶容量 self.tokens capacity self.last_time time.time() self.lock threading.Lock() def consume(self): with self.lock: now time.time() elapsed now - self.last_time self.tokens min(self.capacity, self.tokens elapsed * self.rate) if self.tokens 1: self.tokens - 1 self.last_time now return True return False # 使用示例 bucket TokenBucket(rate0.5, capacity5) # 每秒0.5个请求突发不超过5个 def token_crawl(url): while not bucket.consume(): time.sleep(0.1) return requests.get(url)这个算法允许短时间的突发请求不超过桶容量但长期来看会稳定在设定的速率。比如设置rate0.5表示平均每2秒一个请求capacity5表示最多可以连续发5个请求。4. 高级频率控制策略4.1 基于响应时间的动态调控智能爬虫应该能根据服务器响应时间自动调整请求频率。当响应变慢时说明服务器压力大应该降低请求频率def smart_crawl(urls): avg_response_time 1 # 初始假设1秒响应 for url in urls: start time.time() response requests.get(url) elapsed time.time() - start # 计算移动平均响应时间 avg_response_time 0.8 * avg_response_time 0.2 * elapsed # 动态延迟 平均响应时间 * 安全系数 delay avg_response_time * 2 time.sleep(max(1, delay)) # 不低于1秒这个算法会学习服务器的响应速度始终保持请求间隔是平均响应时间的2倍。当服务器变慢时爬虫会自动降速。4.2 工作日/时段自适应很多网站的负载在不同时间是波动的。我们可以针对不同时段设置不同的爬取策略def time_aware_crawl(url): now datetime.datetime.now() if now.hour in range(9,18): # 工作时间 delay 3 # 白天访问量大延迟高 elif now.hour in range(1,5): # 凌晨 delay 0.5 # 夜间可以加快 else: delay 1 if now.weekday() 5: # 周末 delay * 0.8 # 周末流量低 time.sleep(delay) return requests.get(url)5. 实战注意事项5.1 遵守robots.txt规则正规网站都会在robots.txt中声明爬虫政策。虽然技术上可以无视但道德上应该遵守。使用robotparser模块from urllib import robotparser rp robotparser.RobotFileParser() rp.set_url(https://example.com/robots.txt) rp.read() can_fetch rp.can_fetch(MyBot, https://example.com/page)5.2 请求头伪装技巧合理的请求头能降低被封风险。这是我的常用配置headers { User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36, Accept-Language: zh-CN,zh;q0.9, Referer: https://www.google.com/, Accept-Encoding: gzip, deflate, }特别注意User-Agent要使用常见浏览器版本携带Referer让请求看起来是从搜索引擎跳转来的接受压缩编码减少带宽占用5.3 代理IP轮换策略当需要大规模爬取时单一IP肯定不够用。建议使用代理IP池proxy_pool [ http://proxy1.example.com:8080, http://proxy2.example.com:8080, # ...更多代理 ] def get_with_proxy(url): proxy random.choice(proxy_pool) try: return requests.get(url, proxies{http: proxy}, timeout10) except: print(f代理 {proxy} 失效自动移除) proxy_pool.remove(proxy) return get_with_proxy(url) # 递归重试6. 监控与异常处理完善的爬虫应该具备自我监控能力。这是我的监控方案模板class CrawlerMonitor: def __init__(self): self.stats { total: 0, success: 0, failed: 0, last_error: None, avg_speed: 0, } self.start_time time.time() def record(self, successTrue, errorNone): self.stats[total] 1 if success: self.stats[success] 1 else: self.stats[failed] 1 self.stats[last_error] str(error) # 计算平均速度(请求/分钟) elapsed (time.time() - self.start_time) / 60 self.stats[avg_speed] self.stats[total] / max(1, elapsed) # 自动报警机制 if self.stats[failed] 10 and self.stats[failed] / self.stats[total] 0.2: self.alert_admin() def alert_admin(self): # 实现邮件/短信报警 print(警告爬虫失败率过高)7. 分布式爬虫的限速挑战当爬虫部署在多台机器上时简单的本地限速就不够用了。这时需要分布式限速方案7.1 Redis令牌桶实现import redis import json class DistributedTokenBucket: def __init__(self, host, port, rate, capacity): self.redis redis.Redis(hosthost, portport) self.rate rate self.capacity capacity self.key token_bucket # 初始化桶状态 if not self.redis.exists(self.key): self.redis.set(self.key, json.dumps({ tokens: capacity, last_time: time.time() })) def consume(self): with self.redis.lock(bucket_lock): data json.loads(self.redis.get(self.key)) now time.time() elapsed now - data[last_time] data[tokens] min(self.capacity, data[tokens] elapsed * self.rate) if data[tokens] 1: data[tokens] - 1 data[last_time] now self.redis.set(self.key, json.dumps(data)) return True return False7.2 数据库压力控制对于需要存储爬取结果的场景还要考虑数据库写入压力def save_to_db(data): start time.time() try: # 数据库操作... elapsed time.time() - start # 根据写入速度动态调整 if elapsed 1: # 写入变慢 time.sleep(2) # 延长等待 elif elapsed 0.5: time.sleep(1) except Exception as e: print(f数据库写入失败: {e}) time.sleep(60) # 长时间等待8. 法律与道德考量爬虫开发者必须注意的法律红线不要绕过明显的反爬措施这可能违反计算机犯罪相关法律不要抓取明确声明禁止爬取的数据商业用途抓取前务必咨询法律意见在隐私政策允许的范围内处理个人数据我的个人经验法则是把自己想象成网站管理员思考什么样的爬虫行为是自己能接受的。保持合理的抓取频率给服务器留出足够的响应余量这样既能获取所需数据又不会对目标网站造成负面影响。