文字强调更多
`）需要用 `string()` 函数获取完整文本。 **追问方向**：XPath 1.0 与 2.0 的区别？`normalize-space()` 的作用？ --- ### Q8: CSS Selector 常用语法 **核心答案** ```python 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 中间件开发（完整示例） **核心答案** ```python # 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 中启用中间件：** ```python 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 完整示例 **核心答案** ```python # 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 关键配置说明 **核心答案** | 配置项 | 默认值 | 说明 | | --------------------------------- | ----------------------- | --------------------- | | CONCURRENT\_REQUESTS | 16 | 全局并发请求数 | | CONCURRENT\_REQUESTS\_PER\_DOMAIN | 0（无限制） | 每域名并发数（建议设 2-5） | | DOWNLOAD\_DELAY | 0 | 每次请求间隔（秒） | | RANDOMIZE\_DOWNLOAD\_DELAY | True | 随机化为 0.5\~1.5 倍 | | ROBOTSTXT\_OBEY | True | 是否遵守 robots.txt | | RETRY\_ENABLED | True | 失败重试 | | RETRY\_TIMES | 2 | 最大重试次数 | | RETRY\_HTTP\_CODES | \[500,502,503,504,429\] | 触发重试的状态码 | | HTTPCACHE\_ENABLED | False | 开启可避免重复下载（开发调试用） | | AUTOTHROTTLE\_ENABLED | False | 自动根据服务器响应时间调速 | | AUTOTHROTTLE\_TARGET\_CONCURRENCY | 1.0 | 目标并发数 | | COOKIES\_ENABLED | True | 是否启用 Cookie | | LOG\_LEVEL | DEBUG | 日志级别（生产用 WARNING） | | CLOSESPIDER\_ITEMCOUNT | 0 | 采集到 N 个 Item 后停止（测试用） | | DOWNLOAD\_TIMEOUT | 180 | 下载超时（秒） | | MEMUSAGE\_ENABLED | True | 内存使用监控 | **【注解】** `AUTOTHROTTLE_ENABLED = True` 在生产环境非常有用，可以自动适应服务器负载，避免过快请求导致封禁。`HTTPCACHE_ENABLED = True` 开发阶段开启可大幅节省时间，切记生产关闭。 **追问方向**：AUTOTHROTTLE 的算法原理？如何针对不同 Spider 使用不同配置？ --- ### Q13: Scrapy 去重机制与自定义 **核心答案** Scrapy 默认去重：`RFPDupeFilter`，基于内存 Set 存储请求指纹（URL + 请求方法 + 请求体的 SHA1）。 **自定义布隆过滤器去重（基于 Redis）：** ```python 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 ``` **启用：** ```python # 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 自动链接提取 **核心答案** ```python 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 参数说明：** | 参数 | 类型 | 说明 | | ---------------- | -------- | -------------------------- | | allow | str/list | URL 白名单正则 | | deny | str/list | URL 黑名单正则 | | allow\_domains | list | 允许的域名 | | deny\_domains | list | 排除的域名 | | restrict\_css | str/list | 只在匹配的 CSS 区域内提取链接 | | restrict\_xpaths | str/list | 只在匹配的 XPath 区域内提取链接 | | tags | list | 从哪些标签提取，默认 \['a', 'area'\] | | attrs | list | 从哪些属性提取，默认 \['href'\] | **【注解】** CrawlSpider 不能重写 `parse` 方法（已被框架占用），回调函数必须起其他名字。`restrict_css` 可以避免提取导航栏、页脚等区域的无关链接。 **追问方向**：CrawlSpider 和普通 Spider 如何选择？深度控制如何实现？ --- ## 四、分布式爬虫 ### Q15: scrapy-redis 分布式架构 **核心答案** 核心变更：把内存队列换成 Redis 共享队列，所有节点共享同一 Scheduler 和 DupeFilter，实现多机协同爬取。 **配置：** ```python # 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）：** ```bash 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 对比 **核心答案** | 维度 | Redis Set | 布隆过滤器 | | ----------- | --------- | ------------------------------ | | 内存（1 亿 URL） | \~8 GB | \~200 MB | | 误判率 | 0 | \~0.1%（可调） | | 可删除元素 | 是 | 否（默认） | | 查询复杂度 | O(1) | O(k)，k 为哈希函数个数 | | 实现方式 | Redis SET | Redis BITFIELD 或 RedisBloom 模块 | **选型建议：** - URL 数量 < 1000 万：Redis Set 即可 - URL 数量 > 1000 万：布隆过滤器（接受少量漏爬） - 需要精确去重：Redis Set + 分片 **【注解】** RedisBloom 模块提供 `BF.ADD`/`BF.EXISTS` 命令，比纯 Python 实现的布隆过滤器性能高得多。如果 Redis 服务端不支持 RedisBloom，可以用 Redis bit 数组手动实现（见 Q13、Q34）。 **追问方向**：布隆过滤器的最优哈希函数个数如何推导？ --- ### Q17: Celery 分布式任务调度 **核心答案** ```python # 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：** ```bash celery -A tasks worker --concurrency=4 --loglevel=info # 监控 celery -A tasks flower ``` **【注解】** Celery + Scrapy 集成时，每个任务启动独立的 Scrapy 进程开销较大。高频小任务场景建议直接用 asyncio/aiohttp，Celery 更适合大粒度任务（如每个城市一个任务）。 **追问方向**：Celery 的 chord/chain/group 原语如何使用？如何防止任务重复执行？ --- ## 五、反爬虫对抗 ### Q18: 常见反爬手段与对抗方案 **核心答案** | 反爬手段 | 检测维度 | 对抗方法 | | -------------- | --------------- | ------------------- | | IP 封禁 | 高频同一 IP | 代理池轮换（住宅代理） | | User-Agent 检测 | 非浏览器 UA | 真实 UA + 完整请求头 | | Cookie 验证 | 缺少合法 Cookie | 模拟登录 / Session 维护 | | Referer 检查 | 缺少 Referer | 添加正确 Referer | | Rate Limiting | 高频请求 | 降速 + 随机延迟 | | JS 渲染检测 | 无 JS 执行环境 | Playwright/Selenium | | TLS 指纹（JA3） | 非浏览器 TLS 握手 | curl\_cffi | | 行为分析 | 鼠标/滚动/点击模式 | Playwright 模拟真实行为 | | 验证码 | reCAPTCHA/滑块/图形 | 第三方打码服务/AI | | 蜜罐（Honeypot） | 访问隐藏链接 | 过滤不可见/隐藏元素 | | 设备指纹 | Canvas/WebGL 特征 | Playwright + 注入脚本 | | WebAssembly 加密 | 动态参数签名 | 逆向分析 WASM | **【注解】** 现代网站通常组合使用多种反爬手段，单一对抗往往不够。遇到高强度防护，优先分析 XHR 请求是否有未加密的 API，能直接调接口就不用渲染页面。 **追问方向**：如何检测自己是否被 Cloudflare 拦截？蜜罐的识别方式？ --- ### Q19: 代理 IP 池设计 **核心答案** ```python # 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 与浏览器不同，会被识别为爬虫）。 ```python # 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: 验证码处理方案 **核心答案** ```python 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 逆向基础 **核心答案** ```python # 方案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 页面 **核心答案** ```python 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 反检测配置 **核心答案** ```python 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 | | ----- | ------------------------------- | -------------------------- | | 通信协议 | Chrome DevTools Protocol（CDP）直连 | W3C WebDriver（中间层） | | 异步支持 | 原生 async/await | 需线程池包装 | | 自动等待 | 是（智能等待元素可交互） | 否（需手动 WebDriverWait） | | 网络拦截 | 原生支持 | 需 BrowserMob Proxy 等工具 | | 多浏览器 | Chromium/Firefox/WebKit | Chrome/Firefox/Safari/Edge | | 反检测 | 更好（CDP 层面控制） | 一般（WebDriver 特征明显） | | 执行速度 | 快（CDP 直连） | 慢（多层转发） | | 并发方式 | asyncio + 多 Context | 需多进程/多线程 | | 生态成熟度 | 较新（微软） | 成熟（10+ 年） | **【注解】** 新项目首选 Playwright，异步性能和反检测能力都更强。老项目迁移成本较高时保留 Selenium 也可以。Selenium 4 引入了相对定位等新特性，差距有所缩小。 **追问方向**：Playwright 的 BrowserContext 和 Page 的关系？如何实现并发爬取多个站点？ --- ## 七、异步爬虫 ### Q26: asyncio + aiohttp 并发爬取 **核心答案** ```python 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: 指数退避重试装饰器 **核心答案** ```python 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/PostgreSQL | 事务、索引、SQL 查询灵活 | | 半结构化 JSON、字段不固定 | MongoDB | 无 schema、灵活写入、支持嵌套 | | URL 去重、频率限制、缓存 | Redis | O(1) 操作、持久化可选 | | 全文搜索、日志分析 | Elasticsearch | 倒排索引、聚合分析 | | 大规模离线分析 | Parquet + S3/HDFS | 列式存储、压缩率高、Spark 友好 | | 文件（图片/PDF/视频） | 对象存储（S3/OSS） | 无限扩展、CDN 加速 | **【注解】** 大多数爬虫项目用 MySQL 或 MongoDB 即可。MongoDB 不需要提前定义 schema，爬虫字段经常变化时非常方便，但不适合需要跨集合 join 的场景。URL 去重几乎必用 Redis。 **追问方向**：MongoDB 的 upsert 如何实现？MySQL 分表策略？ --- ### Q29: MySQL 批量插入优化 **核心答案** ```python 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 在爬虫中的作用 **核心答案** ```python 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: 爬取无限滚动页面 **核心答案** ```python 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. 立即止损： ```python # settings.py 临时降速 CONCURRENT_REQUESTS = 1 DOWNLOAD_DELAY = 5 AUTOTHROTTLE_ENABLED = True ``` 3. 对比请求差异（爬虫 vs 浏览器）： ```python 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: 增量爬取设计 **核心答案** ```python 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) ``` **增量爬取策略对比：** | 策略 | 适用场景 | 优点 | 缺点 | | ---------------- | ----------------- | ---------- | --------- | | 基于时间 TTL | 定期更新的页面 | 简单 | 可能爬到相同内容 | | 基于内容哈希 | 内容变化不规律 | 准确 | 需要先下载才能判断 | | 基于 Last-Modified | 支持 HTTP 条件请求的站点 | 服务器配合，流量节省 | 不是所有站点支持 | | 基于站点 Sitemap | 有 sitemap.xml 的站点 | 精确知道更新时间 | 需要站点配合 | **【注解】** 增量爬取的元数据（URL 哈希、最后爬取时间、内容哈希）存 Redis 比存 MySQL 快，但需要设置合理的过期时间防止无限增长。对于新闻类网站，按发布时间过滤往往比内容哈希更高效。 **追问方向**：如何处理 URL 参数顺序不同但内容相同的去重（URL 规范化）？ --- ### Q34: URL 去重系统设计（布隆过滤器） **核心答案** ```python 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('已爬取，跳过') ``` **参数说明：** | 参数 | 类型 | 默认值 | 说明 | | ----------- | ----- | ------------- | ------------- | | capacity | int | 100\_000\_000 | 预期最大元素数量（1 亿） | | error\_rate | float | 0.001 | 允许的误判率（0.1%） | | key | str | — | Redis 中的键名 | **【注解】** `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 过期键 ```python # 流式处理：生成器 + 批量写入 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: 爬虫项目的工程化实践 **核心答案** **日志规范：** ```python 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)) ``` **异常监控：** ```python # 接入 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 的异步引擎。 --- ## 参见 [装饰器与函数高级](https://blog.vercanti.com/python-zhuang-shi-qi-yu-han-shu-gao-ji-yong-fa/) [内置函数完全参考](https://blog.vercanti.com/python-nei-zhi-han-shu-wan-quan-can-kao/) [requests完全指南](https://blog.vercanti.com/requests-wan-quan-zhi-nan/) [httpx完全指南](https://blog.vercanti.com/httpx-wan-quan-zhi-nan/) [asyncio异步编程完全指南](https://blog.vercanti.com/asyncio-yi-bu-bian-cheng-wan-quan-zhi-nan/)