先说结论这个项目让我最深刻的教训是爬虫工程里真正耗时间的从来不是写选择器而是把流程的稳定性焊死。前几天接到一个任务抓取法国FIP展的展商名录。FIP展是巴黎那边一个规模不小的国际性行业展会展商涉及工业设备、原材料、物流服务好几个大类官网结构本身并不花哨但真要完整抓到数据就撞上了三堵墙并发一高就随机报错、国际手机号验证码登录流程繁琐、列表页和详情页的层级关系需要跨页面拼接。这篇文章把这四道难关的攻克过程完整复盘一遍给后续做展会、名录类采集项目的同学当个参照。涉及的核心内容包括Python多线程调度、requests.Session线程安全问题、E.164国际号码规范化、BFS深度爬取策略以及XPath和CSS混合解析二级页面的技巧适合有Python基础、正在做爬虫项目或准备深入数据采集方向的朋友。1. 项目全貌FIP展数据分层与难点成因1.1 数据形态栏目、列表、详情页的三层依赖FIP展官网的信息架构不算复杂结构大概是这样首页有若干个栏目入口比如“参展商”、 “展品分类”、 “展会新闻”。进入“参展商”栏目后是一个典型的列表页带分页每页展示18至24个展商卡片卡片上有公司名、展位号、国家/地区和一个进入详情页的链接。点击链接后才是完整的详情页公司介绍、联系人、电话、邮箱、成立年份、主营品类等字段。数据链路可以用一条简单的依赖链概括栏目页/搜索页 → 列表页含分页 → 详情页难点在于列表页有几百个甚至上千个分页每个分页下又有几十个详情页链接。如果单线程跑一个展商平均1.2秒6000个展商就得跑两个小时以上。业务方又要求当天出完整结果并发是唯一出路。可一旦上并发线程安全问题立刻暴露。与此同时详情页的联系邮箱和电话字段被网站的业务逻辑藏在了“登录后可见”的区域而登录流程强制要求手机短信验证还必须是国际手机号。这就在并发之外又引入了一个独立的登录验证子流程。1.2 四大难点的因果关系这四道关互为因果实际推进时是按这个顺序一个个崩掉的先用单线程脚本跑通了10个展商的流程发现“列表页→详情页”需要用同一个会话保持登录状态开始尝试多线程加速结果触发随机401、请求串号、甚至偶发CSRF token失效这是并发线程安全的第一波爆发并发导致的Session状态错乱又让登录后的验证态时不时丢失被迫回头把国际电话验证流程做成可重试的稳定状态机遍历完列表页之后才发现真正要人工处理的是“详情页URL批量拼接”“分页边界把握”和“二级页面字段的多版本兼容”也就是多页面深度爬取和二级页面解析的常规战役。1.3 技术选型为什么是 requests lxml ThreadPoolExecutor项目初期我对比了几套方案列个表说明当时的取舍方案优势不适合的原因Scrapy自带并发、去重、中间件生态完整学习成本和项目体量不匹配且登录验证流程太定制化框架反而碍手playwright / Selenium能直接扛过前端渲染和复杂验证无头浏览器吃内存并发扩展困难FIP展详情页并非强渲染页面httpx asyncio异步性能好支持HTTP/2项目里大量代码是阻塞式解析和号码验证改造异步要伤筋动骨requests lxml ThreadPoolExecutor简单直接线程模型可控调试方便没有特别短板配合队列后并发表现稳定最终选了第四套。核心思路是requests管HTTPlxml管解析ThreadPoolExecutor管并发queue.Queue管任务调度。这套组合代码写起来直观出问题也好定位。2. 并发线程安全从随机报错到稳定的线程池调度模型2.1 第一个坑多线程共用 requests.Session 导致状态串线业务需求要求登录后抓取所以我一开始很自然地写了一个全局Sessionimport requests session requests.Session() # 全局共享 def fetch_detail_url(url): resp session.get(url, timeout10) return resp.text单线程跑没问题线程一多问题就来了有的线程返回401有的线程拿到的页面里居然混着另一个账号的登录态特征。排查后定位到根因——requests.Session不是为多线程并发设计的。它内部的连接池urllib3 PoolManager和CookieJar都没有做线程间隔离两个线程同时复用同一个TCP连接、同时读写Cookie时轻则连接复用错乱重则请求头和Cookie相互覆盖。解决办法不复杂核心原则是一个线程一个Sessionimport threading import requests thread_local threading.local() def get_session() - requests.Session: if not hasattr(thread_local, session): thread_local.session requests.Session() thread_local.session.headers.update({ User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) ..., Accept-Language: fr-FR,fr;q0.9,en;q0.8, }) return thread_local.session线程本地存储的意思是每个线程第一次调用时创建一个新Session之后这个线程的所有请求都走自己的Session互相不干扰。连接池和Cookie完全隔离保存登录态时也不会串号。2.2 线程池 任务队列不要徒手管理线程用threading.Thread裸创建线程在项目后期是一种灾难。线程生命周期要自己管异常处理要自己包线程数量还没法动态调整。我直接换成了concurrent.futures.ThreadPoolExecutor配合queue.Queue做任务分发。生产者把列表页URL丢进队列消费者线程从队列取任务、抓取列表页、解析出详情页URL后再继续往队列里丢详情页任务。核心结构from queue import Queue from concurrent.futures import ThreadPoolExecutor task_queue Queue() max_workers 8 def producer(): for i in range(1, total_pages 1): task_queue.put({ type: list, url: fhttps://xxxx.fip-expo.com/exhibitors?page{i}, retry: 0 }) def worker(): session get_session() while True: try: task task_queue.get(timeout3) except Exception: break try: if task[type] list: handle_list_page(session, task) else: handle_detail_page(session, task) except Exception as e: log_warning(ftask failed: {task[url]} error{e}) finally: task_queue.task_done() with ThreadPoolExecutor(max_workersmax_workers) as executor: producer() for _ in range(max_workers): executor.submit(worker)队列的线程安全由Python内部保证。队列自带的锁和条件变量天然避免了多个消费者线程抢任务时的竞争。生产者的数量不需要多一个都可以因为消费能力才是瓶颈。2.3 共享计数器的正确写法Lock或原子容器线程池跑起来以后统计成功失败数也成了问题。刚开始我用了一个全局变量success_count 0 failed_count 0结果每次跑完统计数字都对不上这属于典型的“读改写”竞争问题。success_count 1不是原子操作在线程环境里会丢更新。正确做法是加锁或者用线程安全的容器。实际操作里我用了一个带锁的计数器import threading class AtomicCounter: def __init__(self): self._lock threading.Lock() self._value 0 def increment(self, delta1): with self._lock: self._value delta return self._value def value(self): with self._lock: return self._value如果你用的任务都由队列统一管理其实还能走个更省事的路子把计数也变成队列消息由专门的统计线程汇总。但这会让代码结构变重锁的粒度控制好就够用了。2.4 并发参数的实测对比与选择逻辑并发不是越大越好这条经验几乎是老生常谈但具体到自己项目里还是得实测。我在FIP展目标站上分别跑了1、4、8、16、24线程各抓800个展商页面记录耗时和错误率线程数总耗时秒请求失败率备注16120.0%基线慢但稳定4165约0.2%收益明显偶发超时8102约0.8%速度/稳定性平衡点1688约3.1%出现连接被重置2479约5.7%触发服务端限流大量429从数据看8线程是这个项目的最优区间速度接近16线程错误率还能压在1%以内。不要盲目追求高并发“稳定完成”永远比“快那么几秒”重要。3. 国际电话验证从手工登录到工程化短信验证状态机3.1 登录链路梳理申请验证码、回填、建立会话FIP展官网登录强制绑定手机号还要接收短信验证码。它的登录流程是这样的打开登录页输入国际区号手机号点击“发送验证码”服务端给手机发6位数字短信用户输入验证码服务端签发登录态Cookie和Token。在爬虫项目里这五步不能靠人盯着手机输验证码得程序化完成。前提是项目组有合法持有、且在测试环境完成授权的测试号码。整个流程中最麻烦的是第3步和第4步之间的等待与校验以及国际号码的格式标准化。3.2 国际号码标准化法国号码的E.164处理国际化号码最大的坑是格式不统一。同一个号码在页面上可能有这些写法01 40 00 00 00 (0)1 40 00 00 00 0033 1 40 00 00 00 33 1 40 00 00 00针对法国号码规则是先去掉所有非数字字符然后处理前导零。法国本地号码以0开头国际格式要改用33并去掉这个0import re def normalize_fr_phone(raw: str) - str: digits re.sub(r\D, , raw) if digits.startswith(00): digits digits[2:] if digits.startswith(0): digits 33 digits[1:] if not digits.startswith(33): raise ValueError(f无法识别的法国号码格式: {raw}) return digits这样“01 40 00 00 00”就变成“33140000000”可以直接提交到登录接口。国际号码标准化做完以后验证码接收流程才算有了一个可靠入口。3.3 验证码延迟、超时、重复短信的应对机制短信验证码最折磨人的是网络链路长延迟完全不可控。我在项目里用了一个状态机来管理验证码的生命周期IDLE → REQUESTED → RECEIVED → VERIFIED ↓ ↓ FAILED FAILED初始状态是IDLE提交手机号后转到REQUESTED收到短信内容后转到RECEIVED提交验证码成功后转到VERIFIED。任何一步超时都进FAILED并触发重试。等待短信的轮询逻辑实测过后我定成了这样发送验证码请求后先sleep 5秒轮询短信接收结果每次间隔3秒最多20次60秒内未收到发起一次“重新发送验证码”重发后同样等待60秒累积120秒仍未收到整个任务标记失败换号码重试。这个方案的优点是不同时踩雷轮询太频繁容易被服务端当异常行为等待太久又浪费时间。120秒的总体预算是一个平衡点。3.4 会话绑定登录状态必须和线程强绑定登录成功后保存的Session里会带一组Cookie和Token。这里有一个容易被忽视的细节验证码登录完成的Session不能随便让其他线程复用否则就可能出现第2章说的状态串线。实际做法是给每个线程分配固定的“身份槽位”# 一个线程只绑定一个账号会话 session_slot { account_phone: 33140000000, session: get_session(), # thread-local session verified: True, }在这个线程的整个生命周期里所有请求都用这个Session不跨线程传递。这样既绕开了requests.Session的线程安全问题又减少了被风控系统识别为异常共享登录态的概率。4. 多页面深度爬取BFS队列、URL去重与断点续爬4.1 为什么选择BFS而不是DFSFIP展的页面结构是典型的列表页到详情页。遍历策略有DFS和BFS两种实操中我选BFS理由如下BFS保证列表页优先全面覆盖做翻页遍历时能及时发现“某段分页返回500”或“列表结构大改”的问题DFS会一口气钻到底万一详情页解析逻辑坏了整套流程会卡住一半列表页BFS天然配合任务队列列表页任务在前详情页任务在后队列消费顺序可控。任务里带上深度标记task { url: https://xxxx.fip-expo.com/exhibitors?page7, depth: 1, # 列表页 retry: 0, } # 详情页任务 depth2每次从队列取任务先判断depth再走对应的处理函数。这样即使未来网站增加第三级页面改动也局限在任务分发处。4.2 URL去重线程安全只是前提正确性才是关键深度爬取最怕重复抓同一个URL既浪费请求又把数据表整出重复记录。去重集合我放在线程之外共享并且用一个锁保护seen_urls set() seen_lock threading.Lock() def mark_seen(url: str) - bool: with seen_lock: if url in seen_urls: return False seen_urls.add(url) return True当需要更省内存时可以把set换成布隆过滤器。FIP展项目大概有6000-8000个URLset的几百KB内存完全能承受所以没上布隆过滤器。去重时有个细节必须先把URL标准化否则同一个详情页会以“http://”和“https://”两种形式被当成两个URL。统一用urllib.parse.urljoin拼出绝对地址再删掉URL里的锚点片段。4.3 断点续爬内存队列的致命短板线程池跑着跑着网络抖动导致进程异常退出几分钟前队列里积压的任务全部丢失。这是我第一个断点续爬方案踩的坑queue.Queue只在内存里进程一死全没了。后面我把任务状态放进了SQLiteCREATE TABLE crawl_task ( url TEXT PRIMARY KEY, depth INTEGER, status TEXT, -- pending / done / failed retry_count INTEGER, updated_at TEXT );每次把新生成的URL写入crawl_task状态是pending。线程取任务时查询并更新状态def acquire_task(): conn.execute( UPDATE crawl_task SET statusrunning WHERE url IN (SELECT url FROM crawl_task WHERE statuspending LIMIT 1) RETURNING url, depth )任务处理完成后把status改成done。下次启动脚本时只需要查一遍status是pending或running进程崩溃残留的任务重新入队即可。这个改造让整个爬虫变成可重入的不再担心跑到一半宕机。4.4 限速与礼貌抓取别把并发优势跑成劣势多线程快是快但目标站毕竟不是自己家的过度请求会影响别人正常访问。我在每次请求之间加了一个小的随机等待import time import random def polite_sleep(): time.sleep(random.uniform(0.2, 0.5))这个细节让总耗时多了大约十几秒但换来了更低的429错误率。爬虫和站点之间保持基本的默契对双方都好。5. 二级页面解析列表页提链接、详情页提取字段的完整思路5.1 列表页解析从卡片堆里稳定地拿出详情链接列表页通常是一个接一个的展商卡片卡片里有标题、展位号、国家/地区和“查看详情”按钮。我用lxml定位卡片节点再提取详情链接from lxml import html def parse_list_page(html_text, base_url): tree html.fromstring(html_text) cards tree.xpath(//div[contains(class,exhibitor-card)]) for card in cards: rel card.xpath(.//a[classdetail-link]/href) if not rel: continue abs_url urljoin(base_url, rel[0]) yield abs_url列表页解析最容易翻车的地方是卡片CSS类名里的空格和变体。比如exhibitor-card和exhibitor-card wide都能匹配contains(class,exhibitor-card)但如果某一个卡片换了完全不同的类名这一条数据就会静默丢失。所以我每周会抽几个列表页做“结构健康检查”把卡片数量和页面里真正的展商条目数比对不一致就告警。5.2 详情页字段提取XPath CSS 组合拳详情页的字段虽然多但结构相对规整。我的解析函数长这样def parse_detail_page(html_text): tree html.fromstring(html_text) data { name: extract_text(tree, //h1[contains(class,company-name)]), stand_no: extract_text(tree, //div[classstand-info]/span[2]), country: extract_text(tree, //span[contains(class,country)]), category: extract_text(tree, //div[classcategory-tag]), phone: extract_text(tree, //span[itemproptelephone]), email: extract_text(tree, //a[contains(href,mailto)]/href), website: extract_text(tree, //a[classwebsite-link]/href), } data[email] data[email].replace(mailto:, ) return data def extract_text(tree, xpath_expr): nodes tree.xpath(xpath_expr) if nodes and nodes[0].text_content(): return nodes[0].text_content().strip() return XPath负责结构定位CSS选择器负责类名匹配。遇到同一类信息在不同页面有两种结构的情况就用双路选择器互相兜底phone extract_text(tree, //span[itemproptelephone]) if not phone: phone extract_text(tree, //li[contains(class,phone)]/text())5.3 法语内容清洗与字段归一化法语的é、è、à、ç这类重音字符在网页里常被转义成HTML实体比如eacute;。直接落库会出现乱码。解析完成后统一清洗from html import unescape def clean_text(value): if not value: return value unescape(value) # eacute; → é value value.replace(\u200b, ) # 零宽空格 value value.replace(\xa0, ) # 不间断空格 return value.strip()字段归一化还包括电话号码统一去空格和括号邮箱统一转小写网址统一补全协议头。这些琐碎规则能保证最终写进数据库的结果像人手工整理过一样干净。5.4 解析异常的处理静默降级而不是直接抛错详情页偶尔会有缺字段、结构变化甚至整页变成404。我要求解析函数绝不因为单条数据异常拖垮整个流程def safe_parse_detail(html_text): try: return parse_detail_page(html_text) except Exception as e: log_error(fparse failed: {e}) return None # 降级为None后续统一重试解析失败的任务不会直接放弃而是把URL重新塞回队列设置retry计数连续重试三次仍失败才写入专门的失败日志表。这个设计保证了整个爬虫稳定地从第1个展商跑到第6000个展商不会中途卡死。6. 写在最后这四道关教会我的几件事如果让我给这个项目总结成几句个人经验我会这么说第一爬虫并发没有想象中的神秘。线程池加队列是黄金组合比徒手写线程可靠得多。关键是把Session、计数器、任务状态这三样东西的线程安全想清楚别的都可以靠调试解决。第二整个流程里最脆弱的是登录验证环节。它不属于常规的页面解析而是一个偏业务偏网络的子模块。验证码的延迟完全不可控必须用状态机来管理把各种超时和重试路径都覆盖到否则项目每天都在手忙脚乱看日志。第三深度爬取和二级页面解析其实是一对固定搭配。列表页解析负责“找门口”详情页解析负责“进门拿东西”。真正拉开层次的是边界处理——翻页有没有尽头、链接是否重复、字段是否存在、页面结构是否变化。这些问题远比写两个XPath表达式复杂。最后分享一个小技巧并发爬虫里日志一定要带上任务ID或线程ID。这个ID从入队开始跟着任务流转出了问题查日志就是一条完整的链路省下的时间能把排障效率提高一半。这些经验积累下来以后再遇到展会名录、电商商品、新闻聚合这类需要深度遍历的网站心里就有底了。