Python爬虫实战：多线程与异步IO并发爬虫——从10倍到100倍提速

一、为什么你的爬虫那么慢？很多新手写完爬虫后都会遇到同一个问题：单线程爬虫太慢了。你写了一个爬虫，抓取1000个页面要花30分钟。老板说"明天要这些数据"，你直接加班到凌晨。问题出在哪？看这个对比：方式1000个页面耗时提升倍数单线程顺序请求500秒（8分钟）1x多线程（10线程）55秒9x多线程（50线程）15秒33x异步aiohttp（100并发）8秒62x异步+连接池（200并发）5秒100x原因很简单：大部分时间爬虫都在等待服务器响应。网络请求的瓶颈不是你的CPU，而是IO等待。单线程时，发一个请求后程序就傻等，等服务器回数据了再发下一个——这期间CPU完全空闲。并发爬虫的核心思想就是：在等待网络响应的时候，同时发起更多请求。二、最简单的提速：threading多线程先看一个单线程爬虫的典型写法：import requests import time def crawl(url): resp = requests.get(url) return resp.status_code urls = [f"https://httpbin.org/get?id={i}" for i in range(100)] start = time.time() for url in urls: crawl(url) print(f"单线程耗时: {time.time()-start:.2f}秒")改成多线程，只需加几行代码：import requests import time from concurrent.futures import ThreadPoolExecutor, as_completed def crawl(url): resp = requests.get(url, timeout=10) return url, resp.status_code urls = [f"https://httpbin.org/get?id={i}" for i in range(100)] # ========== 多线程版本 ========== start = time.time() with ThreadPoolExecutor(max_workers=20) as executor: futures = {executor.submit(crawl, url): url for url in urls} for future in as_completed(futures): url, status = future.result() print(f"{url} - {status}") print(f"多线程耗时: {time.time()-start:.2f}秒")ThreadPoolExecutor是Python标准库提供的线程池，核心参数就一个：max_workers——同时运行的线程数。线程数怎么选？IO密集型任务（爬虫大部分情况）：20-100线程CPU密集型任务（数据解析如果很复杂）：CPU核心数 × 2目标服务器较弱时：降低线程数，避免被封三、多线程爬虫完整实战：批量采集新闻资讯import requests import time import random import threading from concurrent.futures import ThreadPoolExecutor, as_completed from bs4 import BeautifulSoup import json import os class ConcurrentNewsCrawler: """多线程并发新闻采集器""" def __init__(self, max_workers=20): self.max_workers = max_workers self.results = [] self.lock = threading.Lock() self.session = requests.Session() # 随机UA列表 self.user_agents = [ "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 Chrome/120.0.0.0 Safari/537.36", "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 Chrome/120.0.0.0 Safari/537.36", "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:121.0) Gecko/20100101 Firefox/121.0", "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 Chrome/120.0.0.0 Safari/537.36", ] def get_headers(self): """生成随机请求头""" return { "User-Agent": random.choice(self.user_agents), "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8", "Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8", } def crawl_page(self, url): """抓取单个页面""" try: resp = self.session.get( url, headers=self.get_headers(), timeout=15, ) resp.raise_for_status() resp.encoding = resp.apparent_encoding soup = BeautifulSoup(resp.text, "html.parser") # 提取文章信息（根据实际网站结构调整选择器） item = { "url": url, "title": self._extract_title(soup),