Python爬虫延时案例如何防止爬取过快

wen python案例 2026-06-30 223

本文目录导读：

基础延时方法
自适应延时策略
使用第三方库的多线程延时
线程安全的延时控制
完整的防爬虫延时模块
重要建议

在Python爬虫中防止爬取过快主要涉及延时控制和反反爬策略的结合，以下是详细的延时案例和实现方法：

基础延时方法

time.sleep() - 固定延时

import time
import requests
def crawl_with_delay(url):
    response = requests.get(url)
    time.sleep(3)  # 每次请求后固定等待3秒
    return response.text
# 使用示例
urls = ['http://example.com/page1', 'http://example.com/page2']
for url in urls:
    data = crawl_with_delay(url)
    print(f"抓取: {url}")

random.uniform() - 随机延时

import time
import random
import requests
def crawl_with_random_delay(url):
    # 生成2-5秒之间的随机延时
    delay = random.uniform(2, 5)
    time.sleep(delay)
    response = requests.get(url)
    return response.text

自适应延时策略

基于响应时间的动态延时

import time
import requests
class AdaptiveDelayCrawler:
    def __init__(self, base_delay=1, max_delay=10):
        self.base_delay = base_delay
        self.max_delay = max_delay
        self.last_response_time = 0
    def crawl(self, url):
        start_time = time.time()
        # 根据上次请求耗时调整延时
        adaptive_delay = self.base_delay * (1 + self.last_response_time)
        adaptive_delay = min(adaptive_delay, self.max_delay)
        print(f"等待 {adaptive_delay:.2f} 秒...")
        time.sleep(adaptive_delay)
        response = requests.get(url)
        # 更新响应时间
        self.last_response_time = time.time() - start_time
        return response.text

基于请求成功率的动态调整

import time
import random
import requests
class SmartDelayCrawler:
    def __init__(self):
        self.base_delay = 1
        self.success_count = 0
        self.fail_count = 0
    def crawl(self, url):
        # 根据成功率动态调整延时
        total = self.success_count + self.fail_count
        if total > 10:  # 统计足够样本后再调整
            success_rate = self.success_count / total
            if success_rate < 0.8:  # 成功率低于80%，增加延时
                self.base_delay *= 1.5
            elif success_rate > 0.95:  # 成功率高于95%，减少延时
                self.base_delay *= 0.9
        # 添加随机抖动
        delay = self.base_delay + random.uniform(-0.5, 0.5)
        delay = max(0.5, min(delay, 10))  # 限制在0.5-10秒之间
        time.sleep(delay)
        try:
            response = requests.get(url, timeout=10)
            if response.status_code == 200:
                self.success_count += 1
                return response.text
            else:
                self.fail_count += 1
                print(f"请求失败，状态码: {response.status_code}")
                return None
        except Exception as e:
            self.fail_count += 1
            print(f"请求异常: {e}")
            return None

使用第三方库的多线程延时

使用requests-threads库

from requests_threads import AsyncSession
import time
import random
session = AsyncSession(n=5)  # 设置最大并发数
async def crawl_with_delay(url):
    delay = random.uniform(1, 3)
    await asyncio.sleep(delay)
    response = await session.get(url)
    return response.text
# 批量请求
urls = ['http://example.com/page{}'.format(i) for i in range(20)]
results = session.map(crawl_with_delay, urls)

使用aiohttp的异步延时

import aiohttp
import asyncio
import random
async def crawl_with_delay(session, url):
    delay = random.uniform(1, 3)
    await asyncio.sleep(delay)
    async with session.get(url) as response:
        return await response.text()
async def main():
    urls = ['http://example.com/page{}'.format(i) for i in range(20)]
    async with aiohttp.ClientSession() as session:
        tasks = [crawl_with_delay(session, url) for url in urls]
        results = await asyncio.gather(*tasks)
        return results
# 运行
results = asyncio.run(main())

线程安全的延时控制

使用信号量控制并发

import threading
import time
import requests
class RateLimiter:
    def __init__(self, max_calls=10, period=60):
        self.max_calls = max_calls
        self.period = period
        self.calls = []
        self.lock = threading.Lock()
    def acquire(self):
        with self.lock:
            now = time.time()
            # 清理过期记录
            self.calls = [t for t in self.calls if t > now - self.period]
            if len(self.calls) >= self.max_calls:
                # 计算需要等待的时间
                wait_time = self.calls[0] + self.period - now
                if wait_time > 0:
                    time.sleep(wait_time)
            self.calls.append(time.time())
# 使用示例
rate_limiter = RateLimiter(max_calls=5, period=60)  # 每分钟最多5次请求
def safe_crawl(url):
    rate_limiter.acquire()
    response = requests.get(url)
    return response.text

完整的防爬虫延时模块

import time
import random
import logging
import requests
from functools import wraps
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class AntiCrawlerDelay:
    """综合防爬虫延时控制类"""
    def __init__(self, 
                 min_delay=1, 
                 max_delay=5,
                 retry_delay=10,
                 max_retries=3,
                 enable_jitter=True):
        self.min_delay = min_delay
        self.max_delay = max_delay
        self.retry_delay = retry_delay
        self.max_retries = max_retries
        self.enable_jitter = enable_jitter
    def get_delay(self):
        """获取延时时间"""
        delay = random.uniform(self.min_delay, self.max_delay)
        if self.enable_jitter:
            # 添加随机抖动，增加反检测能力
            jitter = random.uniform(-0.5, 0.5)
            delay = max(self.min_delay, min(delay + jitter, self.max_delay))
        return delay
    def delay_wrapper(self, func):
        """装饰器: 自动添加延时"""
        @wraps(func)
        def wrapper(*args, **kwargs):
            delay = self.get_delay()
            logger.info(f"等待 {delay:.2f} 秒后执行请求...")
            time.sleep(delay)
            for retry in range(self.max_retries):
                try:
                    result = func(*args, **kwargs)
                    return result
                except Exception as e:
                    logger.warning(f"第{retry+1}次请求失败: {e}")
                    if retry < self.max_retries - 1:
                        wait_time = self.retry_delay * (retry + 1)
                        logger.info(f"等待 {wait_time} 秒后重试...")
                        time.sleep(wait_time)
                    else:
                        raise
        return wrapper
# 使用示例
delay_controller = AntiCrawlerDelay(min_delay=2, max_delay=6)
@delay_controller.delay_wrapper
def crawl_page(url):
    response = requests.get(url, timeout=10)
    response.raise_for_status()
    logger.info(f"成功抓取: {url}")
    return response.text
# 批量抓取
urls = ['http://example.com/page{}'.format(i) for i in range(10)]
for url in urls:
    try:
        data = crawl_page(url)
        # 处理数据...
    except Exception as e:
        logger.error(f"抓取 {url} 最终失败: {e}")

重要建议

延时策略选择

简单场景：使用 time.sleep() 或 random.uniform()
高并发场景：使用信号量或异步延时
需要反反爬：结合IP代理、User-Agent轮换和随机延时

最佳实践

import time
import random
from fake_useragent import UserAgent
def advanced_crawl(url):
    # 1. 随机延时
    time.sleep(random.uniform(3, 7))
    # 2. 随机User-Agent
    ua = UserAgent()
    headers = {'User-Agent': ua.random}
    # 3. 设置代理（可选）
    proxies = {
        'http': 'http://proxy.example.com:8080',
        'https': 'https://proxy.example.com:8080'
    }
    # 4. 发送请求
    response = requests.get(url, headers=headers, proxies=proxies)
    return response.text

选择合适的延时策略需要根据实际的反爬虫强度、网络状况和抓取效率要求来综合决定，建议从较慢的速度开始，逐步调整到合适的抓取频率。

本文地址： https://www.bmjup.com/post/12589.html