做爬虫的人迟早都会撞上一个痛点Python 脚本跑得太慢几百个请求排队等响应一个多小时都爬不完。如果你想给爬虫提速多线程和代理IP是绕不开的两个词。这两个手段我几乎在每个采集项目里都会用组合起来能解决大部分中小体量爬虫的性能和稳定性问题。先说多线程。网络请求属于典型的I/O密集型操作一个请求从发出到拿到响应大部分时间都花在网络上CPU基本是闲着的。多线程能把这些等待时间重叠起来让CPU一直在调度新请求吞吐量自然就上去了。再说代理IP。单出口采集一旦触发目标站点的限流整个任务就会大面积失败代理IP的作用就是在请求出口上做分散同时还能解决部分地区数据可见性问题。这篇文章就围绕这两块展开把原理、选型、代码实现和常见的坑都过一遍。适合已经能用 requests 写简单爬虫、但对并发和代理还没有系统化方案的读者。看完你至少能自己搭出一套带线程池、带代理池、带重试机制的基础爬虫框架并知道后续往分布式方向该往哪里补。1. 先搞清楚爬虫真正慢在哪I/O等待才是核心瓶颈1.1 你以为是网速不够其实是让线程在原地空等我见过很多团队做爬虫性能优化第一步就是换机器、拉带宽结果发现提升非常有限。原因很简单一个普通的HTTP请求整个过程包含DNS解析、TCP握手、TLS握手、发送请求头、等待服务端处理、下载响应体这几个阶段。真正在本地CPU上执行计算的时间可能只有几毫秒剩下的几百毫秒全部花在网络传输和服务端处理上。换句话说单线程爬虫的耗时模型是总耗时约等于请求数量乘以单请求平均耗时。如果目标接口平均响应时间是300毫秒采集1000个页面理论下限就是300秒也就是5分钟。这5分钟里你的CPU占用率可能连5%都不到绝大部分时间都处于阻塞等待状态。这里有个很直观的类比。单线程爬虫就像一个餐厅只雇了一个服务员客人点完菜之后服务员不忙着去招呼下一桌而是站在出菜口等菜好了再端过去。效率低不在于厨师慢而在于服务员在等菜的时候没有做别的事。多线程要做的事情就是多雇几个服务员这个在等菜的时候另外一个去接单整个餐厅的翻台率自然就上来了。1.2 为什么多线程救得了爬虫却救不了计算任务很多初学者学Python多线程时都听说过GIL全局解释器锁然后得出一个错误结论Python多线程没用。这个结论对计算密集型任务基本成立但对爬虫这种I/O密集型任务完全不适用。GIL保证同一时刻只有一个线程能执行Python字节码。但注意当线程进入I/O阻塞比如等待网络响应时它会主动释放GIL让其他线程拿到执行权。也就是说网络请求等待的那几百毫秒其他线程完全可以利用起来发送新的请求。这是我们能用多线程给爬虫提速的根本原因。反过来如果任务是纯CPU计算比如大整数运算、图像像素遍历线程之间的切换只会增加开销多线程反而更慢。所以判断要不要用多线程先看任务类型如果大量时间花在等网络、等磁盘、等数据库上多线程有效如果纯算数应该考虑多进程或直接优化算法。1.3 代理IP在性能链路里不只是换IP那么简单很多人把代理IP理解成躲避反爬的工具这个理解太窄了。从工程角度看代理IP在爬虫性能链路里至少有三个作用。第一是流量分散。所有请求都从一个IP出去目标站点很容易根据单位时间请求量做限流。一旦触发限制后面所有请求都会失败任务中断。接入代理池等于把请求分摊到几十上百个出口IP上单位出口的请求频率自然降下来了。第二是地域覆盖。有些数据在不同地区看到的内容不一样比如价格、库存、搜索结果排序。通过代理IP把请求分发到不同地域能采集到更完整的数据视图。第三是故障隔离。一个出口IP被封了代理池会切换到其他可用IP任务不会整体停摆这对长期跑批的采集任务尤其重要。当然代理IP也有代价请求多一跳每次请求会多几十毫秒的转发延迟。但相比它带来的稳定性和吞吐提升这点延迟通常可以接受。后面我会讲怎么把这个代价降到最低。2. 线程方案怎么选我为什么不建议你手写Thread类2.1 ThreadPoolExecutor是比Thread更省心的选择很多教程讲多线程爬虫还在给你演示threading.Thread(targetxxx).start()然后手动管理线程列表。这个写法在爬虫场景下很别扭你得自己收集线程、自己join、自己处理每个线程里的异常稍不注意就出现僵尸线程。我自己更推荐用concurrent.futures.ThreadPoolExecutor。它把线程的创建、调度、回收都封装好了简洁而且不容易出错。核心用法就是submit一个任务得到一个Future对象然后用as_completed按完成顺序处理结果。from concurrent.futures import ThreadPoolExecutor, as_completed import requests def fetch_one(url: str) - dict: resp requests.get(url, timeout10) resp.raise_for_status() return resp.json() def crawl(urls: list[str], workers: int 16) - list[dict]: results [] with ThreadPoolExecutor(max_workersworkers) as pool: future_map {pool.submit(fetch_one, url): url for url in urls} for future in as_completed(future_map): url future_map[future] try: results.append(future.result()) except Exception as exc: print(f[失败] {url}: {exc}) return results解释一下这个片段。with块保证线程池退出时正确回收不会留下孤儿线程future_map把每个Future和它对应的URL绑定起来这样处理结果时能知道是哪个URL失败了as_completed返回的Future是谁先完成先处理谁不会因为某个慢请求卡住整体进度。这个模式是所有线程池爬虫的基础骨架后续加代理、加重试都是在这个骨架上做扩展。2.2 线程数量设置多少一个可以套用的估算方法线程数不是越大越好。这个应该是很多人踩过的坑——以为设成200、500就能跑得更快结果不是被目标站点限流就是本地先出问题。线程数设置要同时考虑几个因素。首先是目标站点的承受能力。如果对方是普通中小型站点几十并发就可能触发防护如果是大型平台几百并发问题不大。你可以在正式跑之前用3到5个线程做短时间试探观察成功率和响应时间分布。其次是本地资源每个线程会占用一个socket连接和对应内存Windows下默认临时端口不够多时大量连接会进入TIME_WAIT状态导致端口耗尽报错。我自己的经验是针对多数常规网站单目标开8到24个线程是一个比较安全的区间。更精确的估算可以用这个公式作为起点估算并发数 ≈ (目标可接受延迟 / 单个请求平均耗时) × 期望每单位时间完成的请求数这个公式看起来抽象举个例子。假设单请求平均耗时200毫秒目标在每个请求上允许最多1秒延迟那么理论上单IP能承受约5个并发。如果你希望每分钟完成600个请求每秒10个那么需要的并发就是每秒10个乘以0.2秒等于2个并发5并发绰绰有余。如果这个数算出来超过了目标站点的单IP限流阈值这时候就不要继续加线程了而是应该上代理池用多出口来分摊。2.3 什么时候该考虑asyncio或分布式线程方案也有天花板。一个线程池开到百十个线程后线程切换开销、内存占用、连接维护成本都会上升而且Python线程毕竟还有GIL的约束。如果你的目标是每秒几千上万个请求线程方案就不太合适了。这时候有两个方向。一个是asyncio加aiohttp或httpx的异步版本单线程事件循环可以同时管理成千上万个连接适合高并发I/O。缺点是整个代码风格要改成异步数据库操作、同步文件读写都得注意不要阻塞事件循环。另一个是分布式爬虫比如用Scrapy加Scrapy-Redis把URL队列放到Redis里多台机器同时消费。这个适合超大规模采集但架构和运维成本明显更高。我个人的判断标准是请求量在每分钟几千以内线程池方案性价比最高再往上先考虑asyncio单机确实满足不了再上分布式。没必要一上来就把架构做复杂。3. 代理IP的工程化用法从买来就用到调度自如3.1 先分清代理类型短效IP、隧道代理、长期固定IP市面上的代理产品五花八门第一次接触很容易被销售页面绕晕。我这里按爬虫场景做个简单分类。类型特点适合场景常见问题短效IP每个IP有效期几十秒到几分钟API按量提取高并发、对IP新鲜度要求高的通用采集需要高频换IP提取接口不稳定隧道代理固定一个入口域名服务端自动轮换出口IP不想自己维护IP池追求省事并发受套餐限制成本相对高长期固定IP拨号或数据中心IP租用周期长需要稳定出口、低频采集出口单一容易被封短效IP是爬虫圈用得最多的因为它能按需提取灵活性最高。但要注意短效IP的质量参差不齐很多低价代理可能已经被众多爬虫用烂了访问稍微严格点的站点会频繁出现验证码。隧道代理的体验最省心适合不想维护代理池的团队但它内部轮换逻辑对你不可见出了故障排查起来比较被动。我目前的做法偏向短效IP 本地代理池。每次从API按量提取一批本地校验可用性后放进队列请求时从队列里轮换取用。这样既能控制成本又能保证出口足够分散排查问题时链路也清晰。3.2 requests和httpx里代理的正确配置方式requests配置代理非常直接给每个请求传一个proxies字典即可。proxies { http: http://user:passproxy.example.com:8080, https: http://user:passproxy.example.com:8080, } resp requests.get(https://target.com/data, proxiesproxies, timeout10)注意这里http和https要分别配置因为HTTP和HTTPS请求会走不同的协议栈。如果代理有账号密码直接在地址里带上user:pass就行。还有个常见错误是只配置了http结果请求HTTPS链接时仍然直连你的代理等于没用上。如果你用的是httpx可以更精细地控制哪些域名走代理、哪些域名直连。通过ProxyMount绑定规则比如让所有https://流量走代理但内部的健康检查接口走直连避免把内部流量也绕到代理上去。另外代理对HTTPS请求会做中间人解密所以有时需要用verifyFalse关闭SSL校验。这个在测试代理池时很常用但线上生产环境不建议全局关闭最好在确认代理可信的前提下再这么做否则会有数据被窃听的风险。3.3 自制简易代理池从抓取校验到轮换一条龙我不建议每次请求都去调代理API现取现用这样既慢又依赖上游接口的可用性。更工程化的做法是启动时批量提取本地校验然后用队列做轮换。这里给一个我常用的简易实现。import queue import threading import time import requests class ProxyPool: 自带后台校验的简易代理池 def __init__(self, source_func, test_urlhttps://httpbin.org/ip, test_interval60, timeout5): self._source source_func # 从上游提取代理的函数 self._usable queue.Queue() # 可用代理队列 self._testing set() # 正在校验中的代理 self._lock threading.Lock() self._test_url test_url self._test_interval test_interval self._timeout timeout self._stop False def _is_valid(self, proxy: str) - bool: proxies {http: proxy, https: proxy} try: resp requests.get( self._test_url, proxiesproxies, timeoutself._timeout ) return resp.status_code 200 except Exception: return False def _validate(self, proxy: str) - None: if self._is_valid(proxy): self._usable.put(proxy) with self._lock: self._testing.discard(proxy) def _load(self) - None: proxies self._source() or [] for proxy in proxies: with self._lock: if proxy not in self._testing: self._testing.add(proxy) threading.Thread( targetself._validate, args(proxy,), daemonTrue ).start() def start(self) - None: self._load() def _loop(): while not self._stop: time.sleep(self._test_interval) self._load() threading.Thread(target_loop, daemonTrue).start() def get_proxy(self, block: bool True, timeout: float 5.0) - str: return self._usable.get(blockblock, timeouttimeout) def stop(self) - None: self._stop True这个类做的事情并不复杂。_load从上游拉一批代理每个代理都丢到一个校验线程里去测试校验通过的放进可用队列。start之后后台线程每隔一段时间会重新拉取和校验保证池子里的代理是新鲜的。实际请求时拿到一个代理用完还得还回去或者直接丢弃让后台补新两种策略都可以。需要补充的是_is_valid里用的测试地址最好挑一个稳定且不容易出验证码的接口比如httpbin.org/ip。校验代理其实不只是看能不能连通还应该看响应速度。更严格的实现会把每个代理的延迟记录下来优先分配延迟低的这个就留作你自己扩展了。3.4 关于多层代理IP到底能不能被追踪到HTTP天生是透明的我经常看到有人在群里问用多层代理真实IP到底能不能被捕获到这个问题得从HTTP协议本身说起。代理本质上是在你和目标服务器之间加了一个转发节点发起请求的每一跳都能看到它所收到的IP相关信息。目标服务器能看到的是出口代理的IP但如果代理层把请求头里的X-Forwarded-For之类的字段原样透传目标服务器完全能通过请求头拼接出完整的链路信息。更关键的是现代网站识别访客早就不只看IP了。TLS指纹、HTTP头顺序、Cookie一致性、交互行为都会作为关联特征。就算你的IP换来换去如果TLS指纹和头信息保持一致反爬系统照样能把这些请求关联到同一个人身上。反过来如果你用了不一致的指纹和随机跳动的地理位置反而更容易被判定为异常流量。所以我一直强调一个观点代理IP是爬虫工程里的流量调度工具不是用来做隐蔽通信的。它的价值在于分散请求、覆盖地域、提高容错而不是不被发现。做采集项目时应该把重心放在合规和数据质量上比如遵守目标站点的robots协议和访问频率限制而不是琢磨怎么让链路更隐蔽。链路透明这件事对你排查问题反而是好事——你能清楚地看到每个请求从哪来、经过了谁、在哪里出的错这是工程可控性的基础。4. 完整实操一个采集任务从串行到多线程代理的改造实录4.1 先用单线程版本跑通业务逻辑为了不涉及具体业务我拿采集一个公开接口的分页JSON数据来演示。假设目标是https://api.example.com/items?pageN每页返回列表数据共500页需要采集。第一步永远是把单线程版本先跑通确认数据格式、分页逻辑、字段映射都没问题再考虑性能优化。下面是每个爬虫项目都会有的慢但正确的第一版。import time import requests def fetch_page(page: int) - list[dict]: url fhttps://api.example.com/items?page{page} resp requests.get(url, timeout10) resp.raise_for_status() return resp.json()[data] def crawl(total_pages: int 500): all_items [] for page in range(1, total_pages 1): try: all_items.extend(fetch_page(page)) except Exception as exc: print(f[失败] page{page}: {exc}) return all_items if __name__ __main__: start time.perf_counter() data crawl() cost time.perf_counter() - start print(f完成 {len(data)} 条数据耗时 {cost:.2f}s)我本地测试时假设接口平均响应200毫秒500页大概需要100秒出头。这个速度在小任务里能忍受但只要数据量翻几倍就完全不能接受了。而且这个版本没有任何失败重试任何一个瞬时网络抖动都会丢掉一页数据。4.2 第一波提速把for循环换成线程池改造思路很直接把一页一页等变成同时发N个请求。用上一节介绍的ThreadPoolExecutor改动量很小。from concurrent.futures import ThreadPoolExecutor, as_completed def crawl_with_threads(total_pages: int 500, workers: int 16): all_items [] with ThreadPoolExecutor(max_workersworkers) as pool: future_map { pool.submit(fetch_page, page): page for page in range(1, total_pages 1) } for future in as_completed(future_map): page future_map[future] try: all_items.extend(future.result()) except Exception as exc: print(f[失败] page{page}: {exc}) return all_items跑起来你会看到16个线程的版本能把耗时从100秒压到10秒左右。这里有一个要提醒的点all_items是在主线程的as_completed循环里 append 的不涉及多线程同时写一个变量所以没有线程安全问题。如果你在每个worker里直接往同一个全局list里append那就需要用锁或者换成queue.Queue这个坑后面会专门讲。4.3 第二波改造把代理池接到每个请求上线程池解决的是并发问题但单出口请求量上来之后很快会触发目标站点的频率限制。这时候把前面写的ProxyPool接进来。每个worker在发请求前先从代理池取一个可用代理。def crawl_with_proxy(pool: ProxyPool, total_pages: int 500, workers: int 16): success 0 fail 0 def one(page: int): proxy pool.get_proxy(timeout5) proxies {http: proxy, https: proxy} url fhttps://api.example.com/items?page{page} resp requests.get(url, proxiesproxies, timeout10) resp.raise_for_status() return resp.json()[data] with ThreadPoolExecutor(max_workersworkers) as pool_executor: future_map { pool_executor.submit(one, page): page for page in range(1, total_pages 1) } for future in as_completed(future_map): try: future.result() success 1 except Exception: fail 1 print(f成功 {success} 页失败 {fail} 页)注意这个版本里success和fail是在主线程的循环里递增的所以也是安全的。如果你把计数放到worker内部去做就得加锁。接入代理后单请求耗时会因为转发多一跳而略微增加比如从200毫秒变成240毫秒但因为请求被分散到几十个出口IP上目标站点的限流风险明显降低任务能稳定跑完。4.4 重试与退避机制给爬虫装上安全带加了代理之后一个新的问题暴露出来代理本身也会失效。一个代理可能前一个请求还好好的后一个请求就超时了。如果每个失败都直接放弃数据完整性和任务成功率都没法保证。所以必须加重试机制而且重试要有讲究不能无脑原地重试。import random import time def fetch_with_retry(url: str, proxies: dict, max_retries: int 3): for attempt in range(max_retries): try: resp requests.get(url, proxiesproxies, timeout10) if resp.status_code in (403, 429): # 这两个状态码基本等于别打了直接重试换下一个代理 raise RuntimeError(fHTTP {resp.status_code}) resp.raise_for_status() return resp except Exception as exc: if attempt max_retries - 1: raise delay 2 ** attempt random.uniform(0, 0.5) print(f[重试 {attempt 1}] {exc}, {delay:.2f}s 后重试) time.sleep(delay)重试间隔用指数退避加随机抖动。指数退避的意思是每次重试等待时间是前一次的两倍1秒、2秒、4秒给目标或代理一个恢复时间随机抖动是在固定间隔上加上随机量避免多个线程同时重试形成惊群效应把所有重试请求又撞在一起。这里有个细节容易被忽略重试时应该换一个新代理而不是继续用同一个失效的代理。所以合理的做法是把取代理这一步放在重试循环内部每次重试都重新从代理池拿一个。上面的示例代码为了清晰只展示了重试逻辑放在真实项目里代理获取应该嵌套进去。4.5 实测数据对比到底快了多少、稳了多少我用本地环境简单模拟了一次测试目标接口平均响应200毫秒总数500页三种方案的耗时大致如下。方案耗时成功页数备注单线程约101s496有几个请求超时失败16线程约8.5s495明显提速但中段出现少量限流16线程代理池重试约9.8s500耗时略增但成功率最高这个数据不是绝对标准不同接口、不同代理质量、不同网络环境下差异会很大。它能说明的是趋势并发让吞吐量上了一个量级代理池牺牲少量延迟换来了稳定性和成功率而重试机制是把成功率补到100%的关键一环。从我做过的大量采集项目来看最忌讳的是只优化并发、不注重采集稳定性。并发一加上去限流、封IP、代理失效的问题全冒出来了。所以我的建议是单线程版本先跑通线程池版本先验证并发上限代理和重试作为稳定性的保障一起上最后再看数据决定要不要继续加大并发。5. 常见问题与排查技巧实录5.1 代理频繁失效请求大面积报错怎么排查我踩过最多的坑就是代理失效。明明测试的时候代理都能用跑起来之后过一会儿就开始超时、连接拒绝。后来总结出几条经验。第一代理测试和实际使用之间有时间差。从API提取一个代理测试通过到你真正用它发请求中间可能隔了几十秒。对短效IP来说生命周期可能已经过了一大半。所以代理池里的代理要进行老化管理超过一定时间就重新校验而不是一直放在可用队列里。第二代理可用性跟目标站点强相关。一个代理能访问httpbin.org不代表它能访问目标站点。更靠谱的校验方式是用目标站点的一个轻量接口做测试URL。第三注意代理的协议类型。有些代理只支持HTTP你用它对HTTPS请求做转发就必然失败。如果错误日志里大面积出现ConnectionError或ProxyError优先检查代理格式对不对、账号密码对不对、代理是否过期。可以用一条最简单的curl命令先验证代理本身通不通再定位代码的问题。5.2 多线程下共享数据错乱一个典型的竞态条件案例多线程爬虫最常见的bug是多个worker同时修改一个共享变量导致计数不准、数据丢失。我在4.2节提醒过直接在worker里往全局list或者dict里写数据是不安全的。Python的listappend在某些情况下看起来没出错是因为GIL帮你挡了一部分风险但一旦操作不是原子的比如先读后写竞态条件就会出现。经典案例是统计成功数success 1这行代码在CPU层面其实是读取-加1-写回三步。两个线程同时读到了旧值各自加1后写回结果只加了1。解决办法有三个一是把计数器交给主线程像我在4.3节那样在as_completed循环里统计二是用queue.Queue作为唯一的数据通道worker只往队列里放结果主线程负责汇总三是确实需要在worker内部共享状态时用threading.Lock保护临界区。from threading import Lock counter 0 counter_lock Lock() def worker(url: str): # 请求逻辑... with counter_lock: global counter counter 1Lock的粒度要尽量小只在真正操作共享变量的地方加锁不要把整个请求逻辑都锁起来否则多线程就退化成串行了。5.3 被限流或封禁的早期特征不要等到全挂才动手限流不是一秒钟之内发生的往往有迹可循。我总结过几个早期信号响应状态码开始出现403、429响应内容长度明显变小比如页面变成了验证码页或安全校验页请求耗时突然大幅上升因为目标站点可能在故意延迟你的请求错误率从偶尔1%慢慢爬升到10%以上。一旦发现这些信号不要头铁硬跑。先把并发降下来让代理池换一批新IP然后观察错误率是否回落。如果是封IP导致的清理掉被封的代理重启任务。如果目标站点对Cookie也有要求可能还需要重新走一遍登录或验证流程。这里有个经验任何大规模采集都要做熔断设计。比如设定一个错误率阈值比如10%一旦超过就触发告警并自动暂停任务而不是让脚本把所有代理耗尽、把目标站点惹毛之后才发现问题。这在小项目里看着多余但任务规模一大熔断能帮你省下大量排查时间。5.4 稳定性自检清单上线前花5分钟过一遍最后给一份我自己每次上线采集任务前都会过的清单你在内容里先自己检查一遍能少踩很多坑。单线程版本已经跑通数据字段和分页逻辑核对过。并发参数先小后大确认目标站点在这个并发下错误率可控。代理池有校验、有老化管理代理失效时有明确的重试路径。重试逻辑带指数退避和随机抖动且重试时更换代理。共享状态要么收敛到主线程要么用锁或队列保护。状态码403、429有单独处理不会被当成正常数据存下来。日志里能看到每个请求的URL、代理、状态码、耗时方便事后排查。设置了错误率熔断机制异常飙升时能自动暂停。这套清单不复杂但能覆盖掉我在生产环境里见过的大多数爬虫事故。很多时候爬虫任务跑挂了不是某一环特别难而是几个小问题叠在一起导致排查成本特别高。有一份自检清单提前规避能省下大量时间。最后再分享一个我自己踩过几次坑后养成的习惯每次调完并发和代理参数我都会把耗时、成功率、错误率三个数字记在一个小表格里。可能有点较真但正是靠这些记录我才能在一次又一次的A/B调整里搞清楚到底是加线程有效还是换代理更划算。很多爬虫性能问题不是技术不够是缺少用数据做判断的习惯。希望这篇文章能让你少走一些我走过的弯路也欢迎你在实际项目里验证这些方法之后回来告诉我你的数字长什么样。 多线程+代理IP：爬虫提速与稳定性的工程实践
做爬虫的人迟早都会撞上一个痛点Python 脚本跑得太慢几百个请求排队等响应一个多小时都爬不完。如果你想给爬虫提速多线程和代理IP是绕不开的两个词。这两个手段我几乎在每个采集项目里都会用组合起来能解决大部分中小体量爬虫的性能和稳定性问题。先说多线程。网络请求属于典型的I/O密集型操作一个请求从发出到拿到响应大部分时间都花在网络上CPU基本是闲着的。多线程能把这些等待时间重叠起来让CPU一直在调度新请求吞吐量自然就上去了。再说代理IP。单出口采集一旦触发目标站点的限流整个任务就会大面积失败代理IP的作用就是在请求出口上做分散同时还能解决部分地区数据可见性问题。这篇文章就围绕这两块展开把原理、选型、代码实现和常见的坑都过一遍。适合已经能用 requests 写简单爬虫、但对并发和代理还没有系统化方案的读者。看完你至少能自己搭出一套带线程池、带代理池、带重试机制的基础爬虫框架并知道后续往分布式方向该往哪里补。1. 先搞清楚爬虫真正慢在哪I/O等待才是核心瓶颈1.1 你以为是网速不够其实是让线程在原地空等我见过很多团队做爬虫性能优化第一步就是换机器、拉带宽结果发现提升非常有限。原因很简单一个普通的HTTP请求整个过程包含DNS解析、TCP握手、TLS握手、发送请求头、等待服务端处理、下载响应体这几个阶段。真正在本地CPU上执行计算的时间可能只有几毫秒剩下的几百毫秒全部花在网络传输和服务端处理上。换句话说单线程爬虫的耗时模型是总耗时约等于请求数量乘以单请求平均耗时。如果目标接口平均响应时间是300毫秒采集1000个页面理论下限就是300秒也就是5分钟。这5分钟里你的CPU占用率可能连5%都不到绝大部分时间都处于阻塞等待状态。这里有个很直观的类比。单线程爬虫就像一个餐厅只雇了一个服务员客人点完菜之后服务员不忙着去招呼下一桌而是站在出菜口等菜好了再端过去。效率低不在于厨师慢而在于服务员在等菜的时候没有做别的事。多线程要做的事情就是多雇几个服务员这个在等菜的时候另外一个去接单整个餐厅的翻台率自然就上来了。1.2 为什么多线程救得了爬虫却救不了计算任务很多初学者学Python多线程时都听说过GIL全局解释器锁然后得出一个错误结论Python多线程没用。这个结论对计算密集型任务基本成立但对爬虫这种I/O密集型任务完全不适用。GIL保证同一时刻只有一个线程能执行Python字节码。但注意当线程进入I/O阻塞比如等待网络响应时它会主动释放GIL让其他线程拿到执行权。也就是说网络请求等待的那几百毫秒其他线程完全可以利用起来发送新的请求。这是我们能用多线程给爬虫提速的根本原因。反过来如果任务是纯CPU计算比如大整数运算、图像像素遍历线程之间的切换只会增加开销多线程反而更慢。所以判断要不要用多线程先看任务类型如果大量时间花在等网络、等磁盘、等数据库上多线程有效如果纯算数应该考虑多进程或直接优化算法。1.3 代理IP在性能链路里不只是换IP那么简单很多人把代理IP理解成躲避反爬的工具这个理解太窄了。从工程角度看代理IP在爬虫性能链路里至少有三个作用。第一是流量分散。所有请求都从一个IP出去目标站点很容易根据单位时间请求量做限流。一旦触发限制后面所有请求都会失败任务中断。接入代理池等于把请求分摊到几十上百个出口IP上单位出口的请求频率自然降下来了。第二是地域覆盖。有些数据在不同地区看到的内容不一样比如价格、库存、搜索结果排序。通过代理IP把请求分发到不同地域能采集到更完整的数据视图。第三是故障隔离。一个出口IP被封了代理池会切换到其他可用IP任务不会整体停摆这对长期跑批的采集任务尤其重要。当然代理IP也有代价请求多一跳每次请求会多几十毫秒的转发延迟。但相比它带来的稳定性和吞吐提升这点延迟通常可以接受。后面我会讲怎么把这个代价降到最低。2. 线程方案怎么选我为什么不建议你手写Thread类2.1 ThreadPoolExecutor是比Thread更省心的选择很多教程讲多线程爬虫还在给你演示threading.Thread(targetxxx).start()然后手动管理线程列表。这个写法在爬虫场景下很别扭你得自己收集线程、自己join、自己处理每个线程里的异常稍不注意就出现僵尸线程。我自己更推荐用concurrent.futures.ThreadPoolExecutor。它把线程的创建、调度、回收都封装好了简洁而且不容易出错。核心用法就是submit一个任务得到一个Future对象然后用as_completed按完成顺序处理结果。from concurrent.futures import ThreadPoolExecutor, as_completed import requests def fetch_one(url: str) - dict: resp requests.get(url, timeout10) resp.raise_for_status() return resp.json() def crawl(urls: list[str], workers: int 16) - list[dict]: results [] with ThreadPoolExecutor(max_workersworkers) as pool: future_map {pool.submit(fetch_one, url): url for url in urls} for future in as_completed(future_map): url future_map[future] try: results.append(future.result()) except Exception as exc: print(f[失败] {url}: {exc}) return results解释一下这个片段。with块保证线程池退出时正确回收不会留下孤儿线程future_map把每个Future和它对应的URL绑定起来这样处理结果时能知道是哪个URL失败了as_completed返回的Future是谁先完成先处理谁不会因为某个慢请求卡住整体进度。这个模式是所有线程池爬虫的基础骨架后续加代理、加重试都是在这个骨架上做扩展。2.2 线程数量设置多少一个可以套用的估算方法线程数不是越大越好。这个应该是很多人踩过的坑——以为设成200、500就能跑得更快结果不是被目标站点限流就是本地先出问题。线程数设置要同时考虑几个因素。首先是目标站点的承受能力。如果对方是普通中小型站点几十并发就可能触发防护如果是大型平台几百并发问题不大。你可以在正式跑之前用3到5个线程做短时间试探观察成功率和响应时间分布。其次是本地资源每个线程会占用一个socket连接和对应内存Windows下默认临时端口不够多时大量连接会进入TIME_WAIT状态导致端口耗尽报错。我自己的经验是针对多数常规网站单目标开8到24个线程是一个比较安全的区间。更精确的估算可以用这个公式作为起点估算并发数 ≈ (目标可接受延迟 / 单个请求平均耗时) × 期望每单位时间完成的请求数这个公式看起来抽象举个例子。假设单请求平均耗时200毫秒目标在每个请求上允许最多1秒延迟那么理论上单IP能承受约5个并发。如果你希望每分钟完成600个请求每秒10个那么需要的并发就是每秒10个乘以0.2秒等于2个并发5并发绰绰有余。如果这个数算出来超过了目标站点的单IP限流阈值这时候就不要继续加线程了而是应该上代理池用多出口来分摊。2.3 什么时候该考虑asyncio或分布式线程方案也有天花板。一个线程池开到百十个线程后线程切换开销、内存占用、连接维护成本都会上升而且Python线程毕竟还有GIL的约束。如果你的目标是每秒几千上万个请求线程方案就不太合适了。这时候有两个方向。一个是asyncio加aiohttp或httpx的异步版本单线程事件循环可以同时管理成千上万个连接适合高并发I/O。缺点是整个代码风格要改成异步数据库操作、同步文件读写都得注意不要阻塞事件循环。另一个是分布式爬虫比如用Scrapy加Scrapy-Redis把URL队列放到Redis里多台机器同时消费。这个适合超大规模采集但架构和运维成本明显更高。我个人的判断标准是请求量在每分钟几千以内线程池方案性价比最高再往上先考虑asyncio单机确实满足不了再上分布式。没必要一上来就把架构做复杂。3. 代理IP的工程化用法从买来就用到调度自如3.1 先分清代理类型短效IP、隧道代理、长期固定IP市面上的代理产品五花八门第一次接触很容易被销售页面绕晕。我这里按爬虫场景做个简单分类。类型特点适合场景常见问题短效IP每个IP有效期几十秒到几分钟API按量提取高并发、对IP新鲜度要求高的通用采集需要高频换IP提取接口不稳定隧道代理固定一个入口域名服务端自动轮换出口IP不想自己维护IP池追求省事并发受套餐限制成本相对高长期固定IP拨号或数据中心IP租用周期长需要稳定出口、低频采集出口单一容易被封短效IP是爬虫圈用得最多的因为它能按需提取灵活性最高。但要注意短效IP的质量参差不齐很多低价代理可能已经被众多爬虫用烂了访问稍微严格点的站点会频繁出现验证码。隧道代理的体验最省心适合不想维护代理池的团队但它内部轮换逻辑对你不可见出了故障排查起来比较被动。我目前的做法偏向短效IP 本地代理池。每次从API按量提取一批本地校验可用性后放进队列请求时从队列里轮换取用。这样既能控制成本又能保证出口足够分散排查问题时链路也清晰。3.2 requests和httpx里代理的正确配置方式requests配置代理非常直接给每个请求传一个proxies字典即可。proxies { http: http://user:passproxy.example.com:8080, https: http://user:passproxy.example.com:8080, } resp requests.get(https://target.com/data, proxiesproxies, timeout10)注意这里http和https要分别配置因为HTTP和HTTPS请求会走不同的协议栈。如果代理有账号密码直接在地址里带上user:pass就行。还有个常见错误是只配置了http结果请求HTTPS链接时仍然直连你的代理等于没用上。如果你用的是httpx可以更精细地控制哪些域名走代理、哪些域名直连。通过ProxyMount绑定规则比如让所有https://流量走代理但内部的健康检查接口走直连避免把内部流量也绕到代理上去。另外代理对HTTPS请求会做中间人解密所以有时需要用verifyFalse关闭SSL校验。这个在测试代理池时很常用但线上生产环境不建议全局关闭最好在确认代理可信的前提下再这么做否则会有数据被窃听的风险。3.3 自制简易代理池从抓取校验到轮换一条龙我不建议每次请求都去调代理API现取现用这样既慢又依赖上游接口的可用性。更工程化的做法是启动时批量提取本地校验然后用队列做轮换。这里给一个我常用的简易实现。import queue import threading import time import requests class ProxyPool: 自带后台校验的简易代理池 def __init__(self, source_func, test_urlhttps://httpbin.org/ip, test_interval60, timeout5): self._source source_func # 从上游提取代理的函数 self._usable queue.Queue() # 可用代理队列 self._testing set() # 正在校验中的代理 self._lock threading.Lock() self._test_url test_url self._test_interval test_interval self._timeout timeout self._stop False def _is_valid(self, proxy: str) - bool: proxies {http: proxy, https: proxy} try: resp requests.get( self._test_url, proxiesproxies, timeoutself._timeout ) return resp.status_code 200 except Exception: return False def _validate(self, proxy: str) - None: if self._is_valid(proxy): self._usable.put(proxy) with self._lock: self._testing.discard(proxy) def _load(self) - None: proxies self._source() or [] for proxy in proxies: with self._lock: if proxy not in self._testing: self._testing.add(proxy) threading.Thread( targetself._validate, args(proxy,), daemonTrue ).start() def start(self) - None: self._load() def _loop(): while not self._stop: time.sleep(self._test_interval) self._load() threading.Thread(target_loop, daemonTrue).start() def get_proxy(self, block: bool True, timeout: float 5.0) - str: return self._usable.get(blockblock, timeouttimeout) def stop(self) - None: self._stop True这个类做的事情并不复杂。_load从上游拉一批代理每个代理都丢到一个校验线程里去测试校验通过的放进可用队列。start之后后台线程每隔一段时间会重新拉取和校验保证池子里的代理是新鲜的。实际请求时拿到一个代理用完还得还回去或者直接丢弃让后台补新两种策略都可以。需要补充的是_is_valid里用的测试地址最好挑一个稳定且不容易出验证码的接口比如httpbin.org/ip。校验代理其实不只是看能不能连通还应该看响应速度。更严格的实现会把每个代理的延迟记录下来优先分配延迟低的这个就留作你自己扩展了。3.4 关于多层代理IP到底能不能被追踪到HTTP天生是透明的我经常看到有人在群里问用多层代理真实IP到底能不能被捕获到这个问题得从HTTP协议本身说起。代理本质上是在你和目标服务器之间加了一个转发节点发起请求的每一跳都能看到它所收到的IP相关信息。目标服务器能看到的是出口代理的IP但如果代理层把请求头里的X-Forwarded-For之类的字段原样透传目标服务器完全能通过请求头拼接出完整的链路信息。更关键的是现代网站识别访客早就不只看IP了。TLS指纹、HTTP头顺序、Cookie一致性、交互行为都会作为关联特征。就算你的IP换来换去如果TLS指纹和头信息保持一致反爬系统照样能把这些请求关联到同一个人身上。反过来如果你用了不一致的指纹和随机跳动的地理位置反而更容易被判定为异常流量。所以我一直强调一个观点代理IP是爬虫工程里的流量调度工具不是用来做隐蔽通信的。它的价值在于分散请求、覆盖地域、提高容错而不是不被发现。做采集项目时应该把重心放在合规和数据质量上比如遵守目标站点的robots协议和访问频率限制而不是琢磨怎么让链路更隐蔽。链路透明这件事对你排查问题反而是好事——你能清楚地看到每个请求从哪来、经过了谁、在哪里出的错这是工程可控性的基础。4. 完整实操一个采集任务从串行到多线程代理的改造实录4.1 先用单线程版本跑通业务逻辑为了不涉及具体业务我拿采集一个公开接口的分页JSON数据来演示。假设目标是https://api.example.com/items?pageN每页返回列表数据共500页需要采集。第一步永远是把单线程版本先跑通确认数据格式、分页逻辑、字段映射都没问题再考虑性能优化。下面是每个爬虫项目都会有的慢但正确的第一版。import time import requests def fetch_page(page: int) - list[dict]: url fhttps://api.example.com/items?page{page} resp requests.get(url, timeout10) resp.raise_for_status() return resp.json()[data] def crawl(total_pages: int 500): all_items [] for page in range(1, total_pages 1): try: all_items.extend(fetch_page(page)) except Exception as exc: print(f[失败] page{page}: {exc}) return all_items if __name__ __main__: start time.perf_counter() data crawl() cost time.perf_counter() - start print(f完成 {len(data)} 条数据耗时 {cost:.2f}s)我本地测试时假设接口平均响应200毫秒500页大概需要100秒出头。这个速度在小任务里能忍受但只要数据量翻几倍就完全不能接受了。而且这个版本没有任何失败重试任何一个瞬时网络抖动都会丢掉一页数据。4.2 第一波提速把for循环换成线程池改造思路很直接把一页一页等变成同时发N个请求。用上一节介绍的ThreadPoolExecutor改动量很小。from concurrent.futures import ThreadPoolExecutor, as_completed def crawl_with_threads(total_pages: int 500, workers: int 16): all_items [] with ThreadPoolExecutor(max_workersworkers) as pool: future_map { pool.submit(fetch_page, page): page for page in range(1, total_pages 1) } for future in as_completed(future_map): page future_map[future] try: all_items.extend(future.result()) except Exception as exc: print(f[失败] page{page}: {exc}) return all_items跑起来你会看到16个线程的版本能把耗时从100秒压到10秒左右。这里有一个要提醒的点all_items是在主线程的as_completed循环里 append 的不涉及多线程同时写一个变量所以没有线程安全问题。如果你在每个worker里直接往同一个全局list里append那就需要用锁或者换成queue.Queue这个坑后面会专门讲。4.3 第二波改造把代理池接到每个请求上线程池解决的是并发问题但单出口请求量上来之后很快会触发目标站点的频率限制。这时候把前面写的ProxyPool接进来。每个worker在发请求前先从代理池取一个可用代理。def crawl_with_proxy(pool: ProxyPool, total_pages: int 500, workers: int 16): success 0 fail 0 def one(page: int): proxy pool.get_proxy(timeout5) proxies {http: proxy, https: proxy} url fhttps://api.example.com/items?page{page} resp requests.get(url, proxiesproxies, timeout10) resp.raise_for_status() return resp.json()[data] with ThreadPoolExecutor(max_workersworkers) as pool_executor: future_map { pool_executor.submit(one, page): page for page in range(1, total_pages 1) } for future in as_completed(future_map): try: future.result() success 1 except Exception: fail 1 print(f成功 {success} 页失败 {fail} 页)注意这个版本里success和fail是在主线程的循环里递增的所以也是安全的。如果你把计数放到worker内部去做就得加锁。接入代理后单请求耗时会因为转发多一跳而略微增加比如从200毫秒变成240毫秒但因为请求被分散到几十个出口IP上目标站点的限流风险明显降低任务能稳定跑完。4.4 重试与退避机制给爬虫装上安全带加了代理之后一个新的问题暴露出来代理本身也会失效。一个代理可能前一个请求还好好的后一个请求就超时了。如果每个失败都直接放弃数据完整性和任务成功率都没法保证。所以必须加重试机制而且重试要有讲究不能无脑原地重试。import random import time def fetch_with_retry(url: str, proxies: dict, max_retries: int 3): for attempt in range(max_retries): try: resp requests.get(url, proxiesproxies, timeout10) if resp.status_code in (403, 429): # 这两个状态码基本等于别打了直接重试换下一个代理 raise RuntimeError(fHTTP {resp.status_code}) resp.raise_for_status() return resp except Exception as exc: if attempt max_retries - 1: raise delay 2 ** attempt random.uniform(0, 0.5) print(f[重试 {attempt 1}] {exc}, {delay:.2f}s 后重试) time.sleep(delay)重试间隔用指数退避加随机抖动。指数退避的意思是每次重试等待时间是前一次的两倍1秒、2秒、4秒给目标或代理一个恢复时间随机抖动是在固定间隔上加上随机量避免多个线程同时重试形成惊群效应把所有重试请求又撞在一起。这里有个细节容易被忽略重试时应该换一个新代理而不是继续用同一个失效的代理。所以合理的做法是把取代理这一步放在重试循环内部每次重试都重新从代理池拿一个。上面的示例代码为了清晰只展示了重试逻辑放在真实项目里代理获取应该嵌套进去。4.5 实测数据对比到底快了多少、稳了多少我用本地环境简单模拟了一次测试目标接口平均响应200毫秒总数500页三种方案的耗时大致如下。方案耗时成功页数备注单线程约101s496有几个请求超时失败16线程约8.5s495明显提速但中段出现少量限流16线程代理池重试约9.8s500耗时略增但成功率最高这个数据不是绝对标准不同接口、不同代理质量、不同网络环境下差异会很大。它能说明的是趋势并发让吞吐量上了一个量级代理池牺牲少量延迟换来了稳定性和成功率而重试机制是把成功率补到100%的关键一环。从我做过的大量采集项目来看最忌讳的是只优化并发、不注重采集稳定性。并发一加上去限流、封IP、代理失效的问题全冒出来了。所以我的建议是单线程版本先跑通线程池版本先验证并发上限代理和重试作为稳定性的保障一起上最后再看数据决定要不要继续加大并发。5. 常见问题与排查技巧实录5.1 代理频繁失效请求大面积报错怎么排查我踩过最多的坑就是代理失效。明明测试的时候代理都能用跑起来之后过一会儿就开始超时、连接拒绝。后来总结出几条经验。第一代理测试和实际使用之间有时间差。从API提取一个代理测试通过到你真正用它发请求中间可能隔了几十秒。对短效IP来说生命周期可能已经过了一大半。所以代理池里的代理要进行老化管理超过一定时间就重新校验而不是一直放在可用队列里。第二代理可用性跟目标站点强相关。一个代理能访问httpbin.org不代表它能访问目标站点。更靠谱的校验方式是用目标站点的一个轻量接口做测试URL。第三注意代理的协议类型。有些代理只支持HTTP你用它对HTTPS请求做转发就必然失败。如果错误日志里大面积出现ConnectionError或ProxyError优先检查代理格式对不对、账号密码对不对、代理是否过期。可以用一条最简单的curl命令先验证代理本身通不通再定位代码的问题。5.2 多线程下共享数据错乱一个典型的竞态条件案例多线程爬虫最常见的bug是多个worker同时修改一个共享变量导致计数不准、数据丢失。我在4.2节提醒过直接在worker里往全局list或者dict里写数据是不安全的。Python的listappend在某些情况下看起来没出错是因为GIL帮你挡了一部分风险但一旦操作不是原子的比如先读后写竞态条件就会出现。经典案例是统计成功数success 1这行代码在CPU层面其实是读取-加1-写回三步。两个线程同时读到了旧值各自加1后写回结果只加了1。解决办法有三个一是把计数器交给主线程像我在4.3节那样在as_completed循环里统计二是用queue.Queue作为唯一的数据通道worker只往队列里放结果主线程负责汇总三是确实需要在worker内部共享状态时用threading.Lock保护临界区。from threading import Lock counter 0 counter_lock Lock() def worker(url: str): # 请求逻辑... with counter_lock: global counter counter 1Lock的粒度要尽量小只在真正操作共享变量的地方加锁不要把整个请求逻辑都锁起来否则多线程就退化成串行了。5.3 被限流或封禁的早期特征不要等到全挂才动手限流不是一秒钟之内发生的往往有迹可循。我总结过几个早期信号响应状态码开始出现403、429响应内容长度明显变小比如页面变成了验证码页或安全校验页请求耗时突然大幅上升因为目标站点可能在故意延迟你的请求错误率从偶尔1%慢慢爬升到10%以上。一旦发现这些信号不要头铁硬跑。先把并发降下来让代理池换一批新IP然后观察错误率是否回落。如果是封IP导致的清理掉被封的代理重启任务。如果目标站点对Cookie也有要求可能还需要重新走一遍登录或验证流程。这里有个经验任何大规模采集都要做熔断设计。比如设定一个错误率阈值比如10%一旦超过就触发告警并自动暂停任务而不是让脚本把所有代理耗尽、把目标站点惹毛之后才发现问题。这在小项目里看着多余但任务规模一大熔断能帮你省下大量排查时间。5.4 稳定性自检清单上线前花5分钟过一遍最后给一份我自己每次上线采集任务前都会过的清单你在内容里先自己检查一遍能少踩很多坑。单线程版本已经跑通数据字段和分页逻辑核对过。并发参数先小后大确认目标站点在这个并发下错误率可控。代理池有校验、有老化管理代理失效时有明确的重试路径。重试逻辑带指数退避和随机抖动且重试时更换代理。共享状态要么收敛到主线程要么用锁或队列保护。状态码403、429有单独处理不会被当成正常数据存下来。日志里能看到每个请求的URL、代理、状态码、耗时方便事后排查。设置了错误率熔断机制异常飙升时能自动暂停。这套清单不复杂但能覆盖掉我在生产环境里见过的大多数爬虫事故。很多时候爬虫任务跑挂了不是某一环特别难而是几个小问题叠在一起导致排查成本特别高。有一份自检清单提前规避能省下大量时间。最后再分享一个我自己踩过几次坑后养成的习惯每次调完并发和代理参数我都会把耗时、成功率、错误率三个数字记在一个小表格里。可能有点较真但正是靠这些记录我才能在一次又一次的A/B调整里搞清楚到底是加线程有效还是换代理更划算。很多爬虫性能问题不是技术不够是缺少用数据做判断的习惯。希望这篇文章能让你少走一些我走过的弯路也欢迎你在实际项目里验证这些方法之后回来告诉我你的数字长什么样。