企业官网建设流程全解析

一、为什么你的爬虫那么慢？

很多新手写完爬虫后都会遇到同一个问题：单线程爬虫太慢了。

你写了一个爬虫，抓取1000个页面要花30分钟。老板说"明天要这些数据"，你直接加班到凌晨。

问题出在哪？看这个对比：

原因很简单：大部分时间爬虫都在等待服务器响应。网络请求的瓶颈不是你的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),