别再瞎配了：爬虫采集器面试真题+完整示例

配置环境就卡半天？依赖冲突、代理失效、IP封禁，这三个坑能劝退90%的新手。今天直接上完整示例，带你拆解高频面试题，代码跑通即掌握。

考点梳理：面试官到底在考什么

别被“采集器”三个字唬住，面试考的不是你会不会写个 requests，而是你对数据获取全链路的理解。核心考点分四层：

1. 基础协议层 HTTP 状态码含义（200/301/302/403/429/503）、Header 伪装（User-Agent/Referer/Cookie）、GET vs POST 参数传递差异。这层是底线，答不上来直接挂。

2. 反爬对抗层 动态渲染（JS 执行）、验证码识别、IP 代理池管理、请求频率控制（Rate Limiting）、TLS 指纹检测。这是区分初级和中级选手的分水岭。

3. 数据清洗层 HTML 解析（XPath/CSS Selector/正则）、去重策略（MD5/布隆过滤器）、结构化存储（JSON/CSV/数据库）。考察你的数据处理能力，而非单纯抓取。

4. 工程化层 异步并发（asyncio/多线程）、错误重试机制、日志监控、分布式部署。这是大厂必问，考察你是否具备生产级思维。

注意：面试官不会只问一个点，而是串联提问。比如：“你遇到过 403 怎么办？”答完 IP 代理后，追问：“代理池怎么维护？”再追问：“如果代理失效率高，怎么优化？”层层递进，答崩了就凉凉。

标准答法：如何组织语言不踩雷

回答时遵循“现象-原因-方案-结果”结构，别背八股文，要讲实战经验。

针对“如何处理反爬”的标准答法：

“我通常分三步走。先看响应码，403 大概率是 Header 或 IP 问题，先换 User-Agent 和代理；如果是 429，说明频率太高，我会在代码里加随机延迟，比如 1-3 秒；如果页面是动态渲染，requests 拿不到数据，我会切到 Playwright 或 Selenium，或者逆向接口直接调 API。每次遇到新站点，我都会先分析 Network 面板，确定是 HTML 还是 JSON，再决定技术栈。”

针对“如何设计高可用采集器”的标准答法：

“核心是隔离和重试。我把请求、解析、存储拆成三个模块，通过消息队列（如 Redis）解耦。请求模块用 asyncio 并发，单个任务失败不影响整体；存储模块加去重逻辑，避免重复入库。监控方面，我记录每次请求的状态码和耗时，超过阈值就告警。比如某次采集电商数据，因 IP 池枯竭导致失败率飙升，我通过监控发现后，动态扩容代理池，20 分钟内恢复稳定。”

避坑指南：

别说“我用爬虫抓数据”，太直白，换成“数据采集”或“信息抽取”。
别吹牛“我能破解所有验证码”，面试官会追问细节，答不上来就是造假。
别提“无限并发”，要强调“合理并发”，体现工程意识。

代码实现：可运行的完整示例

以下是一个基于 asyncio + httpx 的异步采集器，支持代理、重试、去重，可直接运行。参考了官方源码仓库中 httpx 的异步最佳实践。

import asyncio
import hashlib
import json
import random
import time
from typing import Dict, List, Set
import httpxclass WebScraper:def __init__(self, max_concurrency: int = 10, timeout: float = 10.0):self.max_concurrency = max_concurrencyself.timeout = timeoutself.seen_urls: Set[str] = set()self.semaphore = asyncio.Semaphore(max_concurrency)self.proxies = {"http://": "http://proxy1:8080","https://": "http://proxy2:8080"}self.headers = {"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"}def _hash_url(self, url: str) -> str:"""生成 URL 唯一标识，用于去重"""return hashlib.md5(url.encode('utf-8')).hexdigest()async def fetch_with_retry(self, client: httpx.AsyncClient, url: str, retries: int = 3) -> str | None:"""带重试机制的请求- 遇到 429/503 时指数退避- 超过重试次数返回 None"""for attempt in range(retries):try:async with self.semaphore:response = await client.get(url, headers=self.headers, timeout=self.timeout)if response.status_code == 200:return response.textelif response.status_code in [429, 503]:wait_time = (2 ** attempt) + random.uniform(0, 1)print(f"Rate limited, retrying in {wait_time:.2f}s...")await asyncio.sleep(wait_time)else:print(f"Failed to fetch {url}, status: {response.status_code}")return Noneexcept Exception as e:print(f"Request error: {e}, retrying...")await asyncio.sleep(1)return Noneasync def scrape(self, urls: List[str]) -> List[Dict]:"""主采集函数- 并发控制：信号量限制最大并发数- 去重：基于 URL MD5- 异步执行：提升吞吐量"""results = []async with httpx.AsyncClient(proxies=self.proxies) as client:tasks = []for url in urls:url_hash = self._hash_url(url)if url_hash in self.seen_urls:continueself.seen_urls.add(url_hash)tasks.append(self._process_url(client, url))results = await asyncio.gather(*tasks)return [r for r in results if r is not None]async def _process_url(self, client: httpx.AsyncClient, url: str) -> Dict | None:"""处理单个 URL：请求 + 解析"""html = await self.fetch_with_retry(client, url)if not html:return None# 简易解析：提取 <title> 和 <meta description># 实际项目中应使用 BeautifulSoup 或 lxmlimport retitle_match = re.search(r'<title>(.*?)</title>', html, re.DOTALL)desc_match = re.search(r'<meta name="description" content="(.*?)"', html)return {"url": url,"title": title_match.group(1).strip() if title_match else "","description": desc_match.group(1).strip() if desc_match else "","timestamp": time.time()}async def main():scraper = WebScraper(max_concurrency=5)urls = ["https://example.com/page1","https://example.com/page2","https://example.com/page3"]results = await scraper.scrape(urls)print(json.dumps(results, indent=2, ensure_ascii=False))if __name__ == "__main__":asyncio.run(main())

逐行讲解关键点：

asyncio.Semaphore：控制并发数，防止服务器过载，这是生产环境的标配。
hashlib.md5：轻量级去重，适合内存有限的场景。数据量大时换布隆过滤器。
2 ** attempt：指数退避算法，比固定延迟更智能，避免同时重试加剧压力。
httpx.AsyncClient：比 requests 更快，原生支持 HTTP/2，适合高并发场景。

追问与延伸：如何接住连环炮

面试官不会只问一个点，以下高频追问必须准备：

Q1：如果代理 IP 失效率高，怎么办？

“我会建立代理健康检查机制。每次请求前，先 ping 代理 IP，记录成功率。低于 80% 的代理自动踢出池子。同时，接入第三方代理服务商 API，动态获取新 IP。另外，对重要任务采用‘主备代理’策略，主代理失败立即切备用。”

Q2：如何防止被检测到是爬虫？

“三招：一是 Header 伪装，模拟真实浏览器指纹，包括 Accept-Language、Viewport 等；二是行为模拟，加入随机鼠标轨迹、滚动事件，用 Playwright 实现；三是 TLS 指纹，用 curl_cffi 或 undetected-chromedriver 绕过检测。不过最有效的是控制频率，别太贪心。”

Q3：数据量太大，内存放不下怎么办？