Python 爬虫工程师面试题

官方文档：https://docs.python-requests.org/en/latest/
最后更新：2026-05-08

一、HTTP 基础

Q1: HTTP 请求响应完整流程（从输入 URL 到收到响应）

核心答案

DNS 解析 → TCP 三次握手 → TLS 握手（HTTPS）→ 发送 HTTP 请求 → 服务器处理 → 响应 → TCP 挥手

深入分析

1. DNS 解析：浏览器缓存 → 系统 hosts → 本地 DNS → 递归查询权威 DNS
2. TCP 三次握手：SYN → SYN-ACK → ACK，建立可靠连接
3. TLS 握手（HTTPS）：Client Hello → Server Hello + 证书 → 密钥交换 → 握手完成
4. HTTP 请求：请求行 + 请求头 + 请求体
5. 服务器处理：路由 → 业务逻辑 → 数据库 → 响应
6. TCP 四次挥手：FIN → ACK → FIN → ACK，关闭连接（HTTP/1.1 keep-alive 可复用）

爬虫意义：TLS 指纹（JA3）在握手阶段形成，每步都是反爬检测点。

追问方向：TLS 握手细节？HTTP keep-alive 作用？DNS 污染如何处理？

Q2: HTTP/1.1 vs HTTP/2 vs HTTP/3

核心答案

深入分析

• HTTP/2 多路复用：同一 TCP 连接上并发多个请求，消除了 HTTP/1.1 的管道化问题
• HTTP/2 二进制帧：数据以帧（Frame）传输，比文本协议更高效
• HTTP/3 基于 QUIC：UDP 上实现可靠传输，消除 TCP 层的队头阻塞

爬虫实践

# requests 只支持 HTTP/1.1
import requests
resp = requests.get('https://example.com')

# httpx 支持 HTTP/2（需 pip install httpx[http2]）
import httpx
async with httpx.AsyncClient(http2=True) as client:
    resp = await client.get('https://example.com')
    print(resp.http_version)  # HTTP/2

【注解】 大多数爬虫目标使用 HTTP/2，但 requests 始终以 HTTP/1.1 请求，特征明显。对高防护站点建议用 httpx 或 curl_cffi。

追问方向：HTTP/2 多路复用如何实现？QUIC 相比 TCP 的优势？

Q3: Cookie/Session/Token 机制及爬虫处理

核心答案

• Cookie：服务器 Set-Cookie → 浏览器存储 → 后续请求自动携带，状态存客户端
• Session：状态存服务端，通过 session_id（存于 Cookie）关联，requests.Session() 自动管理
• JWT：无状态 Token，Header.Payload.Signature 结构，爬虫需要定期刷新

代码示例：模拟登录保持 Session

import requests

session = requests.Session()

# 第一步：获取 CSRF Token
resp = session.get('https://example.com/login')
csrf = resp.cookies.get('csrftoken')

# 第二步：登录
login_resp = session.post('https://example.com/login', data={
    'username': 'user',
    'password': 'pass',
    'csrfmiddlewaretoken': csrf,
}, headers={'Referer': 'https://example.com/login'})

# 第三步：Session 已维持，直接访问需要登录的页面
data_resp = session.get('https://example.com/protected')

【注解】 CSRF Token 通常嵌在 HTML 表单中或通过接口获取，需要先 GET 登录页再提取。不少网站还校验 Referer，必须加上。

追问方向：JWT 过期如何自动刷新？多账号 Session 并发管理？

Q4: HTTP 状态码与爬虫处理策略

核心答案

【注解】 403 不一定是 IP 封禁，可能只是缺少 Cookie 或 Referer。先检查请求头再换代理，避免浪费代理资源。200 响应体内容也可能是反爬页面（蜜罐），需要校验关键字段是否存在。

追问方向：如何区分 403 是 IP 封禁还是鉴权失败？429 的 Retry-After 头部如何使用？

Q5: requests vs httpx vs aiohttp 对比

核心答案

代码示例：httpx 异步批量请求（HTTP/2）

import httpx
import asyncio

async def fetch_many(urls: list[str]) -> list[dict]:
    results = []
    async with httpx.AsyncClient(http2=True, timeout=10.0) as client:
        tasks = [client.get(url) for url in urls]
        responses = await asyncio.gather(*tasks, return_exceptions=True)
        for url, resp in zip(urls, responses):
            if isinstance(resp, Exception):
                results.append({'url': url, 'error': str(resp)})
            else:
                results.append({'url': url, 'status': resp.status_code})
    return results

urls = ['https://example.com/page/1', 'https://example.com/page/2']
asyncio.run(fetch_many(urls))

【注解】 aiohttp 性能最高但只有异步接口，不熟悉 asyncio 会出很多坑。httpx 兼顾同步/异步，是 requests 的现代替代品。生产环境建议统一用 httpx 或 aiohttp。

追问方向：requests.Session 的连接池参数如何配置？httpx 如何设置重试？

二、HTML 解析

Q6: BeautifulSoup vs lxml vs parsel 对比

核心答案

【注解】 parsel 是 Scrapy 内置解析库，与 Scrapy 的 response.css()/response.xpath() 完全一致，单独使用 parsel 也可以。lxml 解析速度是 BeautifulSoup 的 10-50 倍，大规模项目必用。

追问方向：BeautifulSoup 有哪些解析器（html.parser/lxml/html5lib）？各有什么差异？

Q7: XPath 常用表达式大全

核心答案

from lxml import etree

html = etree.HTML(content)

# 基础选择
html.xpath('//div[@class="item"]')           # class 精确匹配
html.xpath('//div[contains(@class,"item")]') # class 包含
html.xpath('//a/@href')                      # 获取属性值
html.xpath('//span/text()')                  # 获取直接文本
html.xpath('//span//text()')                 # 获取所有后代文本（含子标签）

# 相对路径（在已选节点内继续查找）
node = html.xpath('//div[@id="main"]')[0]
node.xpath('./h2/text()')                    # 直接子节点
node.xpath('.//a/@href')                     # 所有后代 a 标签 href

# 条件与逻辑
html.xpath('//a[contains(@href,"news") and @class="link"]')
html.xpath('//li[position()<=3]')            # 前 3 个
html.xpath('//li[last()]')                   # 最后一个
html.xpath('//li[position() mod 2 = 0]')     # 偶数位置

# 轴（Axis）
html.xpath('//span/parent::div')             # 父节点
html.xpath('//td/following-sibling::td[1]')  # 下一个兄弟
html.xpath('//td/preceding-sibling::td')     # 所有前面的兄弟
html.xpath('//div/ancestor::section')        # 所有祖先 section

# 按文本选取
html.xpath('//button[text()="提交"]')
html.xpath('//div[contains(text(),"关键词")]')
html.xpath('//div[normalize-space(text())="精确匹配去空格"]')

【注解】 text() 只获取当前节点的直接文本子节点，//text() 获取所有后代文本。混合内容（<p>文字<em>强调</em>更多</p>）需要用 string() 函数获取完整文本。

追问方向：XPath 1.0 与 2.0 的区别？normalize-space() 的作用？

Q8: CSS Selector 常用语法

核心答案

from parsel import Selector

sel = Selector(text=html_content)

# 基础选择
sel.css('div.item')                    # class 选择
sel.css('div.item.active')             # 多 class
sel.css('#main-content')               # id 选择
sel.css('div > p')                     # 直接子元素
sel.css('h1 + p')                      # 紧邻兄弟
sel.css('h1 ~ p')                      # 所有后续兄弟

# 属性选择
sel.css('a[href]')                     # 有 href 属性
sel.css('a[href="https://example.com"]') # 精确匹配
sel.css('a[href^="https"]')            # 以 https 开头
sel.css('a[href$=".pdf"]')             # 以 .pdf 结尾
sel.css('a[href*="news"]')             # 包含 news

# 伪类
sel.css('li:first-child')              # 第一个子元素
sel.css('li:last-child')               # 最后一个子元素
sel.css('li:nth-child(2)')             # 第 2 个
sel.css('li:nth-child(odd)')           # 奇数
sel.css('li:not(.disabled)')           # 排除

# 提取内容（parsel 扩展语法）
sel.css('h1::text').get()              # 文本（单个）
sel.css('p::text').getall()            # 文本（全部）
sel.css('a::attr(href)').get()         # 属性（单个）
sel.css('img::attr(src)').getall()     # 属性（全部）

【注解】 ::text 和 ::attr() 是 parsel/Scrapy 的扩展语法，标准 CSS Selector 不支持。get() 返回第一个或 None，getall() 始终返回列表，根据场景选用。

追问方向：CSS Selector 能否替代 XPath？复杂的层级关系哪个更适合？

三、Scrapy 框架深入

Q9: Scrapy 架构与数据流

核心答案

Spider
  ↓ start_requests()
Engine（调度中枢）
  ↓
Scheduler（请求队列，内存/Redis）
  ↓
Engine
  ↓
Downloader Middleware（下载前：UA、代理、签名）
  ↓
Downloader（发出 HTTP 请求）
  ↓
Downloader Middleware（下载后：响应检查、重试）
  ↓
Engine
  ↓
Spider Middleware
  ↓
Spider（parse 方法：提取 Item 和新 Request）
  ↓ yield
Engine
  ↙              ↘
Item Pipeline      Scheduler（新请求入队）
（清洗/存储）

深入分析

• Engine 是核心，连接所有组件，非阻塞事件循环（基于 Twisted）
• Scheduler 管理请求去重（DupeFilter）和优先级队列
• 下载器中间件：process_request / process_response / process_exception 三个钩子
• Spider 中间件：process_spider_output / process_spider_input 两个钩子

【注解】 Scrapy 基于 Twisted 异步框架，不能在 Spider 中使用 time.sleep()，会阻塞整个引擎。需要延迟时使用 DOWNLOAD_DELAY 配置或在中间件中处理。

追问方向：Scrapy 如何实现并发？Twisted 与 asyncio 的关系？

Q10: Scrapy 中间件开发（完整示例）

核心答案

# middlewares.py
import random
import hashlib
import time
import requests as req_lib

# 1. UA 随机轮换中间件
class RandomUserAgentMiddleware:
    USER_AGENTS = [
        'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36',
        'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36',
        'Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36',
    ]

    def process_request(self, request, spider):
        request.headers['User-Agent'] = random.choice(self.USER_AGENTS)


# 2. 代理 IP 中间件
class ProxyMiddleware:
    def __init__(self, proxy_pool_url):
        self.proxy_pool_url = proxy_pool_url

    @classmethod
    def from_crawler(cls, crawler):
        return cls(crawler.settings.get('PROXY_POOL_URL'))

    def process_request(self, request, spider):
        proxy = self._get_proxy()
        if proxy:
            request.meta['proxy'] = proxy

    def process_response(self, request, response, spider):
        if response.status in (403, 429, 503):
            self._mark_proxy_failed(request.meta.get('proxy'))
            request.meta['retry_times'] = request.meta.get('retry_times', 0) + 1
            return request  # 重新调度
        return response

    def _get_proxy(self):
        try:
            resp = req_lib.get(self.proxy_pool_url, timeout=2)
            return resp.text.strip()
        except Exception:
            return None

    def _mark_proxy_failed(self, proxy):
        pass  # 通知代理池标记失效


# 3. 签名请求中间件（添加 token）
class SignatureMiddleware:
    def __init__(self, app_key, app_secret):
        self.app_key = app_key
        self.app_secret = app_secret

    @classmethod
    def from_crawler(cls, crawler):
        return cls(
            crawler.settings.get('APP_KEY'),
            crawler.settings.get('APP_SECRET'),
        )

    def process_request(self, request, spider):
        ts = str(int(time.time()))
        sign_str = f'{self.app_key}{ts}{self.app_secret}'
        sign = hashlib.md5(sign_str.encode()).hexdigest()
        request.headers.update({
            'X-App-Key': self.app_key,
            'X-Timestamp': ts,
            'X-Sign': sign,
        })

settings.py 中启用中间件：

DOWNLOADER_MIDDLEWARES = {
    'myproject.middlewares.RandomUserAgentMiddleware': 400,
    'myproject.middlewares.ProxyMiddleware': 350,
    'myproject.middlewares.SignatureMiddleware': 300,
}

【注解】 中间件的数字表示优先级，数字越小越先执行（process_request 阶段），数字越大越后执行（process_response 阶段则相反）。内置 RetryMiddleware 优先级是 550，自定义重试逻辑要放在它之前（数字更小）。

追问方向：process_request 返回 Response 对象和返回 None 的区别？process_response 可以返回哪些类型？

Q11: Scrapy Item Pipeline 完整示例

核心答案

# pipelines.py
import pymongo
import pymysql
from itemadapter import ItemAdapter
from scrapy.exceptions import DropItem


# 1. 数据清洗 Pipeline
class CleanPipeline:
    def process_item(self, item, spider):
        adapter = ItemAdapter(item)
        for field in ['title', 'content', 'author']:
            if adapter.get(field):
                adapter[field] = adapter[field].strip()
        if not adapter.get('title'):
            raise DropItem(f'标题为空，丢弃: {item.get("url")}')
        return item


# 2. MongoDB 去重写入 Pipeline
class MongoPipeline:
    collection_name = 'articles'

    def __init__(self, mongo_uri, mongo_db):
        self.mongo_uri = mongo_uri
        self.mongo_db = mongo_db

    @classmethod
    def from_crawler(cls, crawler):
        return cls(
            mongo_uri=crawler.settings.get('MONGO_URI'),
            mongo_db=crawler.settings.get('MONGO_DATABASE'),
        )

    def open_spider(self, spider):
        self.client = pymongo.MongoClient(self.mongo_uri)
        self.db = self.client[self.mongo_db]

    def close_spider(self, spider):
        self.client.close()

    def process_item(self, item, spider):
        self.db[self.collection_name].update_one(
            {'url': item['url']},
            {'$set': dict(item)},
            upsert=True,  # 存在则更新，不存在则插入
        )
        return item


# 3. MySQL 批量写入 Pipeline
class MySQLBatchPipeline:
    def __init__(self, mysql_config):
        self.mysql_config = mysql_config
        self.buffer = []
        self.batch_size = 100

    @classmethod
    def from_crawler(cls, crawler):
        return cls(crawler.settings.getdict('MYSQL_CONFIG'))

    def open_spider(self, spider):
        self.conn = pymysql.connect(**self.mysql_config)
        self.cursor = self.conn.cursor()

    def close_spider(self, spider):
        if self.buffer:
            self._flush()
        self.conn.close()

    def process_item(self, item, spider):
        self.buffer.append(dict(item))
        if len(self.buffer) >= self.batch_size:
            self._flush()
        return item

    def _flush(self):
        if not self.buffer:
            return
        sql = 'INSERT IGNORE INTO articles (url, title, content) VALUES (%s, %s, %s)'
        data = [(i['url'], i['title'], i['content']) for i in self.buffer]
        self.cursor.executemany(sql, data)
        self.conn.commit()
        self.buffer.clear()

【注解】 Pipeline 的 process_item 必须返回 item（或抛出 DropItem），否则下游 Pipeline 收到 None 会报错。多个 Pipeline 按 ITEM_PIPELINES 中的数字顺序依次执行。

追问方向：如何让某个 Pipeline 只对特定 Spider 生效？Pipeline 中如何处理异步 IO？

Q12: Scrapy settings.py 关键配置说明

核心答案

【注解】 AUTOTHROTTLE_ENABLED = True 在生产环境非常有用，可以自动适应服务器负载，避免过快请求导致封禁。HTTPCACHE_ENABLED = True 开发阶段开启可大幅节省时间，切记生产关闭。

追问方向：AUTOTHROTTLE 的算法原理？如何针对不同 Spider 使用不同配置？

Q13: Scrapy 去重机制与自定义

核心答案

Scrapy 默认去重：RFPDupeFilter，基于内存 Set 存储请求指纹（URL + 请求方法 + 请求体的 SHA1）。

自定义布隆过滤器去重（基于 Redis）：

import math
import hashlib
import redis
from scrapy.dupefilters import BaseDupeFilter
from scrapy.utils.request import fingerprint


class RedisBloomDupeFilter(BaseDupeFilter):
    """基于 Redis bit 数组的布隆过滤器去重"""

    def __init__(self, redis_url, key='scrapy:bloom',
                 capacity=100_000_000, error_rate=0.001):
        self.r = redis.from_url(redis_url)
        self.key = key
        # 计算最优 bit 数组大小
        self.bit_size = int(-capacity * math.log(error_rate) / (math.log(2) ** 2))
        # 计算最优哈希函数个数
        self.hash_count = int(self.bit_size / capacity * math.log(2))

    @classmethod
    def from_settings(cls, settings):
        return cls(
            redis_url=settings.get('REDIS_URL', 'redis://localhost:6379'),
        )

    def _get_positions(self, fp: str) -> list[int]:
        positions = []
        for i in range(self.hash_count):
            digest = hashlib.sha256(f'{fp}:{i}'.encode()).hexdigest()
            positions.append(int(digest, 16) % self.bit_size)
        return positions

    def request_seen(self, request) -> bool:
        fp = fingerprint(request).hex()
        positions = self._get_positions(fp)
        pipe = self.r.pipeline()
        for pos in positions:
            pipe.getbit(self.key, pos)
        bits = pipe.execute()
        if all(bits):
            return True  # 可能已见过（有误判率）
        # 标记为已见
        pipe = self.r.pipeline()
        for pos in positions:
            pipe.setbit(self.key, pos, 1)
        pipe.execute()
        return False

    def close(self, reason='finished'):
        pass

启用：

# settings.py
DUPEFILTER_CLASS = 'myproject.dupefilters.RedisBloomDupeFilter'
REDIS_URL = 'redis://localhost:6379'

【注解】 默认内存 Set 去重，1 亿 URL 约占 8 GB 内存。布隆过滤器同样规模只需约 200 MB，但有约 0.1% 误判率（误判为已爬，漏掉少量 URL）。分布式爬虫必须用 Redis 去重。

追问方向：布隆过滤器误判率如何计算？如何支持 URL 删除？

Q14: CrawlSpider 自动链接提取

核心答案

from scrapy.spiders import CrawlSpider, Rule
from scrapy.linkextractors import LinkExtractor


class NewsCrawler(CrawlSpider):
    name = 'news'
    allowed_domains = ['example.com']
    start_urls = ['https://example.com/news/']

    rules = (
        # 列表页：匹配分页链接，跟随但不解析 Item
        Rule(
            LinkExtractor(allow=r'/news/page/\d+'),
            follow=True,
        ),
        # 详情页：匹配文章链接，调用 parse_article 解析
        Rule(
            LinkExtractor(allow=r'/news/article/\d+'),
            callback='parse_article',
            follow=False,
        ),
    )

    def parse_article(self, response):
        yield {
            'url': response.url,
            'title': response.css('h1::text').get('').strip(),
            'content': '\n'.join(response.css('.article-body p::text').getall()),
            'pub_date': response.css('.pub-date::text').get(),
        }

LinkExtractor 参数说明：

【注解】 CrawlSpider 不能重写 parse 方法（已被框架占用），回调函数必须起其他名字。restrict_css 可以避免提取导航栏、页脚等区域的无关链接。

追问方向：CrawlSpider 和普通 Spider 如何选择？深度控制如何实现？

四、分布式爬虫

Q15: scrapy-redis 分布式架构

核心答案

核心变更：把内存队列换成 Redis 共享队列，所有节点共享同一 Scheduler 和 DupeFilter，实现多机协同爬取。

配置：

# settings.py
SCHEDULER = 'scrapy_redis.scheduler.Scheduler'
DUPEFILTER_CLASS = 'scrapy_redis.dupefilter.RFPDupeFilter'
REDIS_URL = 'redis://192.168.1.100:6379'
SCHEDULER_PERSIST = True           # 重启后继续爬取，不清空队列
SCHEDULER_QUEUE_CLASS = 'scrapy_redis.queue.PriorityQueue'
# 也可选 SpiderQueue（FIFO）或 SpiderStack（LIFO）

起爬方式（向 Redis 推送起始 URL）：

redis-cli lpush myspider:start_urls '{"url": "https://example.com", "priority": 0}'

架构图：

Master（Redis）
  ├── URL Queue（scrapy_redis:requests）
  └── Seen Set（scrapy_redis:dupefilter）

Worker 1（Scrapy）── 取 URL ──→ 爬取 ──→ 解析 ──→ 新 URL 入队
Worker 2（Scrapy）── 取 URL ──→ 爬取 ──→ 解析 ──→ 新 URL 入队
Worker 3（Scrapy）── 取 URL ──→ 爬取 ──→ 解析 ──→ 新 URL 入队

【注解】 scrapy-redis 本身不处理 Item 存储，仍需要自定义 Pipeline 写入 MongoDB/MySQL。多个 Worker 同时向同一数据库写入时要注意去重（MongoDB upsert 或 MySQL INSERT IGNORE）。

追问方向：如何监控分布式爬虫的进度？Redis 宕机如何处理？

Q16: 布隆过滤器 vs Redis Set 对比

核心答案

选型建议：

• URL 数量 < 1000 万：Redis Set 即可
• URL 数量 > 1000 万：布隆过滤器（接受少量漏爬）
• 需要精确去重：Redis Set + 分片

【注解】 RedisBloom 模块提供 BF.ADD/BF.EXISTS 命令，比纯 Python 实现的布隆过滤器性能高得多。如果 Redis 服务端不支持 RedisBloom，可以用 Redis bit 数组手动实现（见 Q13、Q34）。

追问方向：布隆过滤器的最优哈希函数个数如何推导？

Q17: Celery 分布式任务调度

核心答案

# tasks.py
from celery import Celery

app = Celery('scraper', broker='redis://localhost:6379/0',
             backend='redis://localhost:6379/1')

@app.task(bind=True, max_retries=3, default_retry_delay=60)
def crawl_url(self, url: str, spider_name: str = 'product'):
    """单 URL 爬取任务"""
    try:
        import subprocess
        result = subprocess.run(
            ['scrapy', 'crawl', spider_name, '-a', f'start_url={url}'],
            capture_output=True, text=True, timeout=300,
        )
        if result.returncode != 0:
            raise RuntimeError(result.stderr)
        return {'url': url, 'status': 'success'}
    except Exception as exc:
        raise self.retry(exc=exc)


@app.task
def crawl_batch(urls: list[str]):
    """批量分发爬取任务"""
    from celery import group
    job = group(crawl_url.s(url) for url in urls)
    return job.apply_async()

启动 Worker：

celery -A tasks worker --concurrency=4 --loglevel=info
# 监控
celery -A tasks flower

【注解】 Celery + Scrapy 集成时，每个任务启动独立的 Scrapy 进程开销较大。高频小任务场景建议直接用 asyncio/aiohttp，Celery 更适合大粒度任务（如每个城市一个任务）。

追问方向：Celery 的 chord/chain/group 原语如何使用？如何防止任务重复执行？

五、反爬虫对抗

Q18: 常见反爬手段与对抗方案

核心答案

【注解】 现代网站通常组合使用多种反爬手段，单一对抗往往不够。遇到高强度防护，优先分析 XHR 请求是否有未加密的 API，能直接调接口就不用渲染页面。

追问方向：如何检测自己是否被 Cloudflare 拦截？蜜罐的识别方式？

Q19: 代理 IP 池设计

核心答案

# proxy_pool.py
import redis
import requests
import time


class ProxyPool:
    """基于 Redis Sorted Set 的代理池，分值越高质量越好"""

    def __init__(self, redis_url: str = 'redis://localhost:6379'):
        self.r = redis.from_url(redis_url)
        self.key = 'proxy:pool'
        self.min_score = 20.0       # 低于此分值淘汰
        self.max_score = 100.0      # 最高分值

    def add(self, proxy: str, score: float = 100.0):
        self.r.zadd(self.key, {proxy: score})

    def get(self) -> str | None:
        """获取分值最高的可用代理"""
        proxies = self.r.zrevrangebyscore(
            self.key, '+inf', self.min_score, start=0, num=1
        )
        return proxies[0].decode() if proxies else None

    def decrease_score(self, proxy: str, amount: float = 10.0):
        score = self.r.zscore(self.key, proxy)
        if score is None:
            return
        new_score = score - amount
        if new_score < self.min_score:
            self.r.zrem(self.key, proxy)
        else:
            self.r.zadd(self.key, {proxy: new_score})

    def increase_score(self, proxy: str, amount: float = 5.0):
        score = self.r.zscore(self.key, proxy)
        if score is None:
            return
        self.r.zadd(self.key, {proxy: min(score + amount, self.max_score)})

    def validate_all(self):
        """定时验证所有代理，恢复有效代理的分值"""
        proxies = self.r.zrange(self.key, 0, -1)
        for proxy_bytes in proxies:
            proxy = proxy_bytes.decode()
            if self._check(proxy):
                self.r.zadd(self.key, {proxy: self.max_score})
            else:
                self.decrease_score(proxy, 50)

    def _check(self, proxy: str, test_url: str = 'https://httpbin.org/ip') -> bool:
        try:
            resp = requests.get(
                test_url,
                proxies={'http': proxy, 'https': proxy},
                timeout=5,
            )
            return resp.status_code == 200
        except Exception:
            return False

    @property
    def size(self) -> int:
        return self.r.zcard(self.key)

【注解】 免费代理质量极差，生产环境必须用付费代理（机房代理或住宅代理）。住宅代理 IP 来自真实用户设备，反检测效果远好于机房 IP，但价格更贵。代理池需要配合定时任务（如 APScheduler）持续补充和验证代理。

追问方向：机房代理 vs 住宅代理 vs 移动代理的区别？如何选择代理供应商？

Q20: TLS 指纹（JA3）与 curl_cffi

核心答案

JA3 指纹：TLS Client Hello 消息中的 TLS 版本、密码套件、扩展列表、椭圆曲线等字段的 MD5 哈希，可唯一标识客户端（requests 的 JA3 与浏览器不同，会被识别为爬虫）。

# requests 的 JA3 与浏览器不同，会被高防站点识别
import requests
resp = requests.get('https://tls.browserleaks.com/json')
print(resp.json())  # ja3_hash 与浏览器不同

# curl_cffi：使用 libcurl 并模拟浏览器的 TLS 指纹
from curl_cffi import requests as cffi_requests

resp = cffi_requests.get(
    'https://tls.browserleaks.com/json',
    impersonate='chrome120',  # 模拟 Chrome 120 的完整 TLS 指纹
)
print(resp.json())  # ja3_hash 与真实 Chrome 一致

# 支持的 impersonate 值：
# chrome99, chrome100, ..., chrome120
# firefox99, firefox100, ...
# safari15, safari16, ...
# edge99, edge101, ...

# 异步支持
from curl_cffi.requests import AsyncSession

async def fetch(url):
    async with AsyncSession() as session:
        resp = await session.get(url, impersonate='chrome120')
        return resp.text

【注解】 curl_cffi 是绕过 Cloudflare、Akamai 等基于 TLS 指纹检测的最有效工具。如果 curl_cffi 也被检测，通常是行为特征（请求频率、时序）问题，需要结合 Playwright。

追问方向：JA3 指纹如何计算？除了 TLS 指纹还有哪些指纹特征？

Q21: 验证码处理方案

核心答案

import ddddocr
import requests
import time


# 方案1：ddddocr 识别图形验证码（本地 AI）
def solve_image_captcha(img_url: str) -> str:
    ocr = ddddocr.DdddOcr(show_ad=False)
    img_data = requests.get(img_url).content
    return ocr.classification(img_data)


# 方案2：2Captcha 解决 reCAPTCHA v2（第三方服务）
def solve_recaptcha_v2(site_key: str, page_url: str, api_key: str) -> str:
    # 提交任务
    submit = requests.post('https://2captcha.com/in.php', data={
        'key': api_key,
        'method': 'userrecaptcha',
        'googlekey': site_key,
        'pageurl': page_url,
        'json': 1,
    }).json()

    if submit['status'] != 1:
        raise RuntimeError(f'提交失败: {submit}')

    task_id = submit['request']

    # 轮询结果（最多等 100 秒）
    for _ in range(20):
        time.sleep(5)
        result = requests.get('https://2captcha.com/res.php', params={
            'key': api_key,
            'action': 'get',
            'id': task_id,
            'json': 1,
        }).json()
        if result['status'] == 1:
            return result['request']
        if result['request'] != 'CAPCHA_NOT_READY':
            raise RuntimeError(f'验证码失败: {result}')

    raise TimeoutError('验证码解决超时')


# 方案3：Playwright 处理滑块验证码（模拟人工拖动）
async def solve_slider_captcha(page, slider_selector: str, track_selector: str):
    slider = await page.query_selector(slider_selector)
    track = await page.query_selector(track_selector)

    slider_box = await slider.bounding_box()
    track_box = await track.bounding_box()

    start_x = slider_box['x'] + slider_box['width'] / 2
    start_y = slider_box['y'] + slider_box['height'] / 2
    end_x = track_box['x'] + track_box['width'] - 10

    # 模拟人类拖动（非匀速，带随机抖动）
    await page.mouse.move(start_x, start_y)
    await page.mouse.down()
    steps = 30
    for i in range(steps):
        x = start_x + (end_x - start_x) * (i + 1) / steps
        y = start_y + ((-1) ** i) * 2  # 轻微上下抖动
        await page.mouse.move(x, y)
        await page.wait_for_timeout(10 + (i % 5) * 3)
    await page.mouse.up()

【注解】 ddddocr 对简单英数字图形验证码识别率约 80-90%，复杂的需要训练自定义模型。第三方打码服务（2Captcha/Anti-Captcha）成本低但有延迟（10-30秒）。滑块验证码的轨迹需要模拟加速-匀速-减速的人类行为，纯匀速会被识别。

追问方向：reCAPTCHA v3 如何处理（基于行为评分）？行为式验证码（如腾讯天御）的对抗思路？

Q22: JavaScript 逆向基础

核心答案

# 方案1：execjs 执行 JS 代码（简单场景）
import execjs
import hashlib


js_code = """
function generateSign(paramsStr, timestamp, secret) {
    return CryptoJS.MD5(paramsStr + timestamp + secret).toString();
}
"""

# 加载 CryptoJS 库 + 业务代码
ctx = execjs.compile(open('crypto-js.min.js').read() + js_code)

params = {'user_id': '123', 'page': '1'}
params_str = '&'.join(f'{k}={v}' for k, v in sorted(params.items()))
sign = ctx.call('generateSign', params_str, '1234567890', 'secret_key')


# 方案2：Python 重写加密逻辑（推荐，性能更好）
def python_sign(params: dict, timestamp: str, key: str) -> str:
    sorted_str = '&'.join(f'{k}={v}' for k, v in sorted(params.items()))
    sign_str = f'{sorted_str}{timestamp}{key}'
    return hashlib.md5(sign_str.encode()).hexdigest()


# 方案3：用 Node.js 子进程执行复杂 JS
import subprocess
import json


def call_js(js_file: str, function_name: str, *args) -> str:
    args_json = json.dumps(args)
    script = f"""
    const func = require('./{js_file}').{function_name};
    const args = {args_json};
    console.log(JSON.stringify(func(...args)));
    """
    result = subprocess.run(
        ['node', '-e', script],
        capture_output=True, text=True, timeout=10,
    )
    return json.loads(result.stdout.strip())

【注解】 execjs 每次调用都有启动开销，高频调用性能差。复杂的 JS 环境（如有 DOM 依赖）应该用 Node.js 子进程或 Playwright 执行。能用 Python 重写的加密逻辑一定要重写，可维护性和性能都更好。逆向时重点关注：网络请求的 sign/token 参数从哪里来，用 Chrome Network + Sources 面板追踪调用栈。

追问方向：如何找到加密函数的入口？混淆 JS（如 OB 混淆）如何还原？详见 js逆向/README。

六、JavaScript 渲染页面

Q23: Playwright 异步爬取 SPA 页面

核心答案

from playwright.async_api import async_playwright
import asyncio


async def scrape_spa(url: str) -> str:
    async with async_playwright() as p:
        browser = await p.chromium.launch(
            headless=True,
            args=['--no-sandbox', '--disable-dev-shm-usage'],
        )
        context = await browser.new_context(
            user_agent='Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36',
            viewport={'width': 1920, 'height': 1080},
            locale='zh-CN',
        )
        page = await context.new_page()

        # 拦截不必要的资源（加速 20-40%）
        await page.route(
            '**/*.{png,jpg,gif,svg,woff,woff2,ttf,mp4}',
            lambda route: route.abort()
        )

        await page.goto(url, wait_until='networkidle', timeout=30000)
        await page.wait_for_selector('.article-content', timeout=10000)

        content = await page.inner_text('.article-content')
        await browser.close()
        return content


# 拦截 XHR/fetch 直接获取 API 数据（无需解析 HTML）
async def intercept_api(url: str) -> list:
    api_data = []

    async with async_playwright() as p:
        browser = await p.chromium.launch(headless=True)
        page = await browser.new_page()

        async def handle_response(response):
            if 'api/items' in response.url and response.status == 200:
                try:
                    data = await response.json()
                    api_data.extend(data.get('items', []))
                except Exception:
                    pass

        page.on('response', handle_response)
        await page.goto(url, wait_until='domcontentloaded')
        await page.wait_for_timeout(3000)
        await browser.close()

    return api_data


# 并发多页面爬取
async def crawl_many_spa(urls: list[str], concurrency: int = 5) -> list[str]:
    sem = asyncio.Semaphore(concurrency)

    async def fetch_one(url):
        async with sem:
            return await scrape_spa(url)

    return await asyncio.gather(*[fetch_one(url) for url in urls])

【注解】 wait_until='networkidle' 等待所有网络请求完成，最稳但最慢。动态页面推荐用 wait_for_selector() 等待关键元素。拦截网络请求直接获取 JSON 数据比解析渲染后的 HTML 更高效，应优先考虑。

追问方向：Playwright 如何复用浏览器上下文（Context）？如何处理弹窗、文件下载？

Q24: Playwright 反检测配置

核心答案

from playwright.async_api import async_playwright


async def create_stealth_browser():
    async with async_playwright() as p:
        browser = await p.chromium.launch(
            headless=True,
            args=[
                '--no-sandbox',
                '--disable-blink-features=AutomationControlled',  # 关键：移除自动化标志
            ],
        )
        context = await browser.new_context(
            user_agent='Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36',
            viewport={'width': 1920, 'height': 1080},
        )

        # 注入反检测脚本（在每个页面加载前执行）
        await context.add_init_script("""
            // 移除 webdriver 标志
            Object.defineProperty(navigator, 'webdriver', {
                get: () => undefined,
            });

            // 伪造 chrome 对象（非无头模式下存在）
            window.chrome = {
                runtime: {},
                loadTimes: function() {},
                csi: function() {},
                app: {},
            };

            // 修正 plugins 长度（无头模式为 0，正常浏览器 > 0）
            Object.defineProperty(navigator, 'plugins', {
                get: () => [1, 2, 3, 4, 5],
            });

            // 修正语言
            Object.defineProperty(navigator, 'languages', {
                get: () => ['zh-CN', 'zh', 'en'],
            });

            // 防止 headless 检测
            Object.defineProperty(navigator, 'platform', {
                get: () => 'Win32',
            });
        """)

        return browser, context

【注解】 playwright-stealth 库（第三方）封装了更完整的反检测脚本，可直接使用。Playwright 默认 headless 模式会暴露 navigator.webdriver = true，必须手动处理。--disable-blink-features=AutomationControlled 是最重要的启动参数。

追问方向：如何检测自己的爬虫是否被识别为自动化工具（访问 bot.sannysoft.com）？

Q25: Playwright vs Selenium 对比

核心答案

【注解】 新项目首选 Playwright，异步性能和反检测能力都更强。老项目迁移成本较高时保留 Selenium 也可以。Selenium 4 引入了相对定位等新特性，差距有所缩小。

追问方向：Playwright 的 BrowserContext 和 Page 的关系？如何实现并发爬取多个站点？

七、异步爬虫

Q26: asyncio + aiohttp 并发爬取

核心答案

import asyncio
import aiohttp
from asyncio import Semaphore


async def fetch(
    session: aiohttp.ClientSession,
    url: str,
    sem: Semaphore,
) -> dict:
    async with sem:  # 限制并发数，防止过载
        try:
            async with session.get(
                url,
                timeout=aiohttp.ClientTimeout(total=10),
            ) as resp:
                resp.raise_for_status()
                text = await resp.text()
                return {'url': url, 'status': resp.status, 'content': text}
        except aiohttp.ClientError as e:
            return {'url': url, 'error': str(e)}
        except asyncio.TimeoutError:
            return {'url': url, 'error': 'timeout'}


async def crawl_many(urls: list[str], concurrency: int = 20) -> list[dict]:
    sem = Semaphore(concurrency)

    # 连接池：最大 100 个连接，禁用 SSL 验证（加速，内网使用）
    connector = aiohttp.TCPConnector(limit=100, ssl=False)
    timeout = aiohttp.ClientTimeout(total=30, connect=5)

    async with aiohttp.ClientSession(
        connector=connector,
        timeout=timeout,
        headers={'User-Agent': 'Mozilla/5.0 ...'},
    ) as session:
        tasks = [fetch(session, url, sem) for url in urls]
        return await asyncio.gather(*tasks, return_exceptions=True)


# 使用
urls = [f'https://example.com/page/{i}' for i in range(1000)]
results = asyncio.run(crawl_many(urls, concurrency=30))
success = [r for r in results if isinstance(r, dict) and 'content' in r]

【注解】 asyncio.gather 的 return_exceptions=True 很重要，否则任何一个任务异常都会导致整批任务失败。并发数不是越大越好，20-50 通常是合理范围，过大会被目标服务器封禁，也会耗尽本地端口。

追问方向：aiohttp 和 asyncio 的关系？如何实现带速率限制的生产者-消费者模式？

Q27: 指数退避重试装饰器

核心答案

import asyncio
import functools
import random
import logging


def async_retry(
    max_retries: int = 3,
    base_delay: float = 1.0,
    max_delay: float = 60.0,
    exceptions: tuple = (Exception,),
):
    """异步函数指数退避重试装饰器"""
    def decorator(func):
        @functools.wraps(func)
        async def wrapper(*args, **kwargs):
            last_error = None
            for attempt in range(max_retries + 1):
                try:
                    return await func(*args, **kwargs)
                except exceptions as e:
                    last_error = e
                    if attempt == max_retries:
                        break
                    # 指数退避：1s → 2s → 4s → ...
                    delay = min(base_delay * (2 ** attempt), max_delay)
                    # 加入随机抖动（防止惊群效应）
                    jitter = random.uniform(0, delay * 0.1)
                    wait_time = delay + jitter
                    logging.warning(
                        f'{func.__name__} 第 {attempt + 1} 次失败: {e}，'
                        f'{wait_time:.1f}s 后重试'
                    )
                    await asyncio.sleep(wait_time)
            raise last_error
        return wrapper
    return decorator


# 使用示例
import aiohttp

@async_retry(max_retries=3, base_delay=2.0, exceptions=(aiohttp.ClientError, asyncio.TimeoutError))
async def fetch_with_retry(url: str) -> str:
    async with aiohttp.ClientSession() as session:
        async with session.get(url, timeout=aiohttp.ClientTimeout(total=10)) as resp:
            resp.raise_for_status()
            return await resp.text()

【注解】 随机抖动（jitter）是防止"惊群效应"的关键——如果大量请求同时失败，没有抖动的话它们会在完全相同的时间点同时重试，造成第二次冲击。指数退避配合抖动是分布式系统的标准做法。

追问方向：同步版本的重试装饰器如何实现？tenacity 库有什么优势？

八、数据存储

Q28: 爬虫存储方案选型

核心答案

【注解】 大多数爬虫项目用 MySQL 或 MongoDB 即可。MongoDB 不需要提前定义 schema，爬虫字段经常变化时非常方便，但不适合需要跨集合 join 的场景。URL 去重几乎必用 Redis。

追问方向：MongoDB 的 upsert 如何实现？MySQL 分表策略？

Q29: MySQL 批量插入优化

核心答案

import pymysql
import time

conn = pymysql.connect(
    host='localhost', db='crawler',
    charset='utf8mb4',
    cursorclass=pymysql.cursors.DictCursor,
)

items = [
    ('https://example.com/1', '标题1', '内容1', time.time()),
    ('https://example.com/2', '标题2', '内容2', time.time()),
]

with conn.cursor() as cursor:
    # 方案1：INSERT IGNORE（主键/唯一索引冲突时忽略，不报错）
    sql = '''
        INSERT IGNORE INTO articles (url, title, content, crawl_time)
        VALUES (%s, %s, %s, %s)
    '''
    affected = cursor.executemany(sql, items)
    print(f'插入 {affected} 条')

    # 方案2：ON DUPLICATE KEY UPDATE（冲突时更新指定字段）
    sql2 = '''
        INSERT INTO articles (url, title, content, crawl_time)
        VALUES (%s, %s, %s, %s)
        ON DUPLICATE KEY UPDATE
            title = VALUES(title),
            content = VALUES(content),
            crawl_time = VALUES(crawl_time)
    '''
    cursor.executemany(sql2, items)

conn.commit()

# 性能对比（10000 条数据）：
# 循环 execute：约 10s
# executemany：约 1s（10x 提升）
# LOAD DATA INFILE：约 0.1s（100x 提升，适合超大量）

【注解】 executemany 会将多条数据合并为一条 INSERT 语句（VALUES(...),(...),(...) 形式），减少网络往返次数。url 字段必须建唯一索引才能使用 INSERT IGNORE 或 ON DUPLICATE KEY UPDATE。批量大小建议 100-500 条，过大会导致内存占用过高。

追问方向：如何估算合适的批量大小？asyncio 环境下如何使用 MySQL（aiomysql）？

Q30: Redis 在爬虫中的作用

核心答案

import redis
import time

r = redis.Redis(decode_responses=True)

# 1. URL 任务队列（List）
# 生产者：入队
r.lpush('url:queue', 'https://example.com/page/1', 'https://example.com/page/2')
# 消费者：阻塞出队（最多等待 5 秒）
item = r.brpop('url:queue', timeout=5)
if item:
    _, url = item

# 2. 去重集合（Set）
url = 'https://example.com/article/123'
if r.sismember('url:seen', url):
    print('已爬取，跳过')
else:
    r.sadd('url:seen', url)
    # 开始爬取...

# 3. 代理池（Sorted Set，分值=质量分）
r.zadd('proxy:pool', {'http://1.2.3.4:8080': 100.0})
best_proxies = r.zrevrange('proxy:pool', 0, 4)  # 取分值最高的 5 个

# 4. 请求频率限制（滑动窗口）
def check_rate_limit(domain: str, limit: int = 10, window: int = 60) -> bool:
    """判断 domain 在 window 秒内是否超过 limit 次请求"""
    key = f'rate:{domain}:{int(time.time() // window)}'
    count = r.incr(key)
    if count == 1:
        r.expire(key, window)  # 第一次设置过期时间
    return count <= limit

# 5. 分布式锁（防止多 Worker 重复爬取同一 URL）
import hashlib

def crawl_with_lock(url: str):
    url_hash = hashlib.md5(url.encode()).hexdigest()[:8]
    lock_key = f'lock:url:{url_hash}'
    acquired = r.set(lock_key, '1', nx=True, ex=30)  # nx=仅不存在时设置，ex=30秒超时
    if not acquired:
        return None  # 其他 Worker 正在处理
    try:
        # 执行爬取
        pass
    finally:
        r.delete(lock_key)  # 释放锁

# 6. 爬取进度统计（Hash）
r.hset('crawl:stats', mapping={
    'total': 1000,
    'done': 0,
    'failed': 0,
})
r.hincrby('crawl:stats', 'done', 1)

【注解】 Redis 分布式锁的 ex 超时时间要大于单次爬取的最大耗时，否则任务未完成锁就过期，造成重复爬取。brpop 的阻塞模式比轮询节省 CPU，是任务队列的标准做法。

追问方向：Redis 分布式锁的 Redlock 算法是什么？Redis 持久化（RDB vs AOF）如何配置？

九、综合场景题

Q31: 爬取无限滚动页面

核心答案

import requests
import time


# 方案1：分析 AJAX 请求，直接调用接口（推荐）
def scrape_infinite_scroll_api(max_pages: int = 10) -> list:
    """通过 Chrome Network 面板分析 XHR 请求，直接调用分页 API"""
    results = []
    headers = {
        'X-Requested-With': 'XMLHttpRequest',
        'Referer': 'https://example.com/list',
    }
    for page in range(1, max_pages + 1):
        resp = requests.get(
            'https://example.com/api/items',
            params={'page': page, 'size': 20, '_t': int(time.time())},
            headers=headers,
        )
        data = resp.json()
        items = data.get('items') or data.get('data') or []
        if not items:
            break
        results.extend(items)
        time.sleep(1)  # 礼貌延迟
    return results


# 方案2：Playwright 模拟滚动
from playwright.async_api import async_playwright

async def scroll_and_scrape(url: str) -> list:
    async with async_playwright() as p:
        browser = await p.chromium.launch(headless=True)
        page = await browser.new_page()
        await page.goto(url)

        all_items = []
        prev_count = 0
        max_scrolls = 50

        for _ in range(max_scrolls):
            # 滚动到底部
            await page.evaluate('window.scrollTo(0, document.body.scrollHeight)')
            await page.wait_for_timeout(2000)  # 等待新内容加载

            current_items = await page.query_selector_all('.item')
            if len(current_items) == prev_count:
                break  # 没有新数据，停止
            prev_count = len(current_items)

        # 提取所有数据
        for item in current_items:
            title = await item.query_selector('.title')
            all_items.append({
                'title': await title.inner_text() if title else '',
            })

        await browser.close()
        return all_items

【注解】 优先分析 Network 面板找到 API 直接调用，效率是 Playwright 的 10 倍以上。很多无限滚动页面的 API 有 cursor/offset/page 参数，找到规律就能无需渲染直接爬取。只有真正找不到 API 时再用 Playwright 滚动。

追问方向：cursor 分页和 page 分页的区别？如何处理分页 API 需要 token 的情况？

Q32: 爬虫被封后的应急处理

核心答案

诊断步骤：

1. 确认封禁类型：

   • 换代理能访问 → IP 封禁
   • 换代理仍然不行 → Cookie/账号封禁或设备指纹
   • 返回 JS 挑战页面 → Cloudflare/盾类防护
   • 返回 200 但内容异常 → 蜜罐响应

2. 立即止损：

   # settings.py 临时降速
   CONCURRENT_REQUESTS = 1
   DOWNLOAD_DELAY = 5
   AUTOTHROTTLE_ENABLED = True
   

3. 对比请求差异（爬虫 vs 浏览器）：

   import requests
   
   # 检查目标站点能看到的请求头
   resp = requests.get(
       'https://httpbin.org/headers',
       headers={
           'User-Agent': 'Mozilla/5.0 ...',
           'Accept': 'text/html,application/xhtml+xml,...',
           'Accept-Language': 'zh-CN,zh;q=0.9',
           'Accept-Encoding': 'gzip, deflate, br',
           'Connection': 'keep-alive',
       }
   )
   print(resp.json())
   

4. 处理方案优先级：

   • IP 封禁：切换住宅代理池
   • Cookie 失效：重新模拟登录获取 Session
   • TLS 指纹：改用 curl_cffi（impersonate='chrome120'）
   • 行为检测：降速 + 随机延迟 + Playwright 模拟真实交互
   • 验证码：接入第三方打码服务

【注解】 被封后不要急着换代理，先确认封禁类型。很多时候只是请求头不完整（缺少 Accept、Accept-Language 等），补全请求头比换代理更有效。记录封禁时间点和当时的并发数，有助于找到安全阈值。

追问方向：如何设计爬虫的自动熔断机制（连续失败超过阈值自动暂停）？

Q33: 增量爬取设计

核心答案

import hashlib
import time
import redis


class IncrementalCrawler:
    def __init__(self, redis_url: str = 'redis://localhost:6379'):
        self.r = redis.from_url(redis_url, decode_responses=True)

    def _url_key(self, url: str) -> str:
        url_hash = hashlib.md5(url.encode()).hexdigest()
        return f'crawl:{url_hash}'

    def should_recrawl(
        self,
        url: str,
        content: str = None,
        ttl: int = 86400,  # 24 小时
    ) -> bool:
        """判断是否需要重新爬取"""
        key = self._url_key(url)

        # 策略1：基于时间（TTL 内不重爬）
        last_crawl = self.r.hget(key, 'time')
        if last_crawl and time.time() - float(last_crawl) < ttl:
            return False

        # 策略2：基于内容哈希（内容未变化则跳过）
        if content is not None:
            content_hash = hashlib.md5(content.encode()).hexdigest()
            stored_hash = self.r.hget(key, 'hash')
            if stored_hash == content_hash:
                # 更新最后检查时间，但标记内容未变
                self.r.hset(key, 'time', time.time())
                return False
            self.r.hset(key, 'hash', content_hash)

        self.r.hset(key, 'time', time.time())
        self.r.expire(key, ttl * 7)  # 元数据保留 7 倍 TTL
        return True

    def mark_failed(self, url: str, error: str):
        key = self._url_key(url)
        self.r.hset(key, mapping={
            'error': error,
            'fail_time': time.time(),
        })
        self.r.hincrby(key, 'fail_count', 1)

增量爬取策略对比：

【注解】 增量爬取的元数据（URL 哈希、最后爬取时间、内容哈希）存 Redis 比存 MySQL 快，但需要设置合理的过期时间防止无限增长。对于新闻类网站，按发布时间过滤往往比内容哈希更高效。

追问方向：如何处理 URL 参数顺序不同但内容相同的去重（URL 规范化）？

Q34: URL 去重系统设计（布隆过滤器）

核心答案

import math
import hashlib
import redis


class BloomFilter:
    """基于 Redis bit 数组的布隆过滤器，适用于 URL 去重"""

    def __init__(
        self,
        redis_client: redis.Redis,
        key: str,
        capacity: int = 100_000_000,
        error_rate: float = 0.001,
    ):
        self.r = redis_client
        self.key = key
        # 最优 bit 数组大小
        self.bit_size = int(-capacity * math.log(error_rate) / (math.log(2) ** 2))
        # 最优哈希函数个数
        self.hash_count = max(1, int(self.bit_size / capacity * math.log(2)))

    def _get_positions(self, item: str) -> list[int]:
        """计算 item 对应的 k 个 bit 位置"""
        positions = []
        for i in range(self.hash_count):
            digest = hashlib.sha256(f'{item}:{i}'.encode()).hexdigest()
            positions.append(int(digest, 16) % self.bit_size)
        return positions

    def add(self, item: str):
        """添加元素"""
        pipe = self.r.pipeline()
        for pos in self._get_positions(item):
            pipe.setbit(self.key, pos, 1)
        pipe.execute()

    def contains(self, item: str) -> bool:
        """检查元素是否存在（可能误判，不会漏判）"""
        pipe = self.r.pipeline()
        for pos in self._get_positions(item):
            pipe.getbit(self.key, pos)
        return all(pipe.execute())

    def add_and_check(self, item: str) -> bool:
        """原子性地检查并添加，返回是否已存在"""
        pipe = self.r.pipeline()
        positions = self._get_positions(item)
        for pos in positions:
            pipe.getbit(self.key, pos)
        bits = pipe.execute()
        is_seen = all(bits)
        if not is_seen:
            pipe = self.r.pipeline()
            for pos in positions:
                pipe.setbit(self.key, pos, 1)
            pipe.execute()
        return is_seen

    @property
    def memory_usage_mb(self) -> float:
        """估算内存占用（MB）"""
        return self.bit_size / 8 / 1024 / 1024


# 使用
r = redis.Redis()
bf = BloomFilter(r, key='url:bloom', capacity=100_000_000, error_rate=0.001)
print(f'内存占用约: {bf.memory_usage_mb:.1f} MB')  # ~179.5 MB

url = 'https://example.com/article/123'
if not bf.add_and_check(url):
    print('新 URL，开始爬取')
else:
    print('已爬取，跳过')

参数说明：

【注解】 add_and_check 不是严格原子性的（两个 pipeline 之间有窗口），高并发下极小概率出现竞争。分布式场景下可用 Lua 脚本保证原子性，或接受极低概率的漏判（对爬虫场景通常可接受）。

追问方向：如何计算布隆过滤器的最优参数？误判率与容量、哈希函数数量的关系？

十、附加题

Q35: 爬虫项目的性能调优经验

核心答案

CPU 侧：

• 使用 asyncio + aiohttp 或 httpx 代替同步请求，IO 等待期间可处理其他任务
• 解析密集型（lxml）可用 ProcessPoolExecutor 利用多核

IO 侧：

• 数据库批量写入（executemany），避免逐条插入
• Redis Pipeline 批量命令，减少网络往返
• 连接池复用（aiohttp.TCPConnector、pymysql.connect 池化）

网络侧：

• 拦截无关资源（图片、字体），减少带宽消耗和等待时间
• 合理设置超时（connect_timeout=3s，read_timeout=10s）
• 开启 HTTP keep-alive 复用连接

内存侧：

• 使用生成器（yield）流式处理数据，不要把所有结果加载到内存
• 布隆过滤器代替 Set 去重
• 定期清理 Redis 过期键

# 流式处理：生成器 + 批量写入
def process_items(spider_output):
    buffer = []
    for item in spider_output:
        buffer.append(item)
        if len(buffer) >= 100:
            db.bulk_insert(buffer)
            buffer.clear()
    if buffer:
        db.bulk_insert(buffer)

【注解】 性能优化要先定位瓶颈（是网络 IO、CPU 解析还是数据库写入），再针对性优化，避免盲目优化。使用 cProfile 或 py-spy 分析 CPU 热点，使用 aiomonitor 监控协程状态。

Q36: 爬虫项目的工程化实践

核心答案

日志规范：

import logging
import json
from datetime import datetime

def setup_logging(spider_name: str):
    logging.basicConfig(
        level=logging.INFO,
        format='%(asctime)s [%(levelname)s] %(name)s: %(message)s',
        handlers=[
            logging.StreamHandler(),
            logging.FileHandler(f'logs/{spider_name}.log', encoding='utf-8'),
        ],
    )

# 结构化日志（便于 ELK 分析）
def log_item(logger, url: str, status: str, duration: float):
    logger.info(json.dumps({
        'event': 'item_crawled',
        'url': url,
        'status': status,
        'duration': duration,
        'timestamp': datetime.utcnow().isoformat(),
    }, ensure_ascii=False))

异常监控：

# 接入 Sentry
import sentry_sdk
sentry_sdk.init(dsn='your-dsn', traces_sample_rate=0.1)

# 自定义告警（钉钉/飞书/邮件）
def alert_on_failure(spider_name: str, error: str, threshold: int = 100):
    """连续失败超过阈值时发送告警"""
    key = f'alert:fail:{spider_name}'
    count = r.incr(key)
    if count == 1:
        r.expire(key, 3600)
    if count >= threshold:
        send_alert(f'{spider_name} 连续失败 {count} 次: {error}')
        r.delete(key)

目录结构：

project/
├── spiders/          # Spider 文件
├── middlewares/      # 中间件
├── pipelines/        # Pipeline
├── items.py          # Item 定义
├── settings.py       # 配置
├── utils/            # 工具函数（签名、解密等）
├── tests/            # 单元测试
├── logs/             # 日志
└── requirements.txt  # 依赖

【注解】 爬虫项目容易忽视工程化，上线后定位问题非常困难。最低要求：结构化日志 + 关键指标监控（成功率、QPS、数据量）+ 异常告警。建议用 Prometheus + Grafana 监控爬虫运行状态。

最佳实践

回答架构题时先画分层结构：爬虫系统设计题（调度、去重、存储、反爬）先在草稿上画出数据流，口述时按"数据从哪来 → 怎么处理 → 存到哪"的顺序展开，避免跳跃。面试官评分看的是结构化思维，而非记住所有细节。

准备真实项目数字：面试时被问"你的爬虫规模"时要有具体数字：日抓取页面数、平均延迟、成功率、去重规模。若没有生产数据，用压测结果替代，避免只说"很快""很稳定"等模糊描述。

反爬应对题突出合规前提：被问如何绕过验证码/IP 封锁时，先说明在授权范围内（如自有数据、学术研究、明确允许爬取的场景），再介绍技术手段（代理池、随机 UA、请求间隔），体现职业素养。

并发模型按场景选择并能说清 Trade-off：asyncio 适合 IO 密集（单线程高并发）、concurrent.futures.ThreadPoolExecutor 适合混合任务、multiprocessing 适合 CPU 密集（如 HTML 解析 + JS 渲染）。面试中说清楚"为什么选这个"比说出实现细节更重要。

分布式题先提单机瓶颈：被问"如何做分布式爬虫"时，先说明什么情况下单机不够（带宽、速率限制、目标站点 IP 分散），再引出任务队列（Redis/Kafka）+ 多 Worker 模式，而不是直接上架构，让面试官看到你的工程判断力。

常见陷阱

陷阱：回答"如何提升爬取速度"只提并发数

现象： 面试官追问"还有什么办法？"时无法继续。
原因： 速度瓶颈不只在并发数，还有 DNS 查询、TCP 连接建立、响应解析、数据写入、目标服务器限速等环节。
解决： 回答时分层：网络层（keep-alive 复用连接、requests.Session）、协议层（HTTP/2 多路复用、gzip 压缩）、解析层（lxml 比 BeautifulSoup 快 10x）、存储层（批量写入而非逐条插入）。

陷阱：说"用 time.sleep 避免被封"但忽略随机化

现象： 面试官问"固定间隔有什么问题？"时答不上来。
原因： 固定间隔产生规律性请求模式，反爬系统容易检测；真实浏览器行为的请求间隔是随机分布的。
解决： 用 time.sleep(random.uniform(1, 3)) 加随机抖动；高频场景结合请求频率监控（滑动窗口限速），而不是简单 sleep。

陷阱：混淆同步与异步 requests 库

现象： 被问"你怎么做异步爬取"时回答"用 requests 的异步模式"，但 requests 本身是同步库。
原因： requests 不支持 asyncio；异步 HTTP 客户端是 httpx（带 AsyncClient）或 aiohttp。
解决： 明确区分：requests 用于同步/线程池场景，httpx.AsyncClient 或 aiohttp.ClientSession 用于 asyncio 场景；Scrapy 有自己基于 Twisted 的异步引擎。

参见

装饰器与函数高级
内置函数完全参考
requests完全指南
httpx完全指南
asyncio异步编程完全指南