全栈攻城狮

Appearance

爬虫基础与 Java/Python 爬虫框架 ​

爬虫基础 ​

HTTP 请求 ​

HTTP 请求是爬虫与目标服务器通信的基础。爬虫通过模拟浏览器发送 HTTP 请求获取服务器响应的 HTML、JSON 或其他格式的数据。

GET 请求

GET 请求用于从服务器获取资源。参数通过 URL 的查询字符串传递。

python
import requests

# GET 请求示例
response = requests.get('https://api.example.com/users', params={'page': 1, 'limit': 20})
print(response.status_code)
print(response.text)
java
import java.net.URI;
import java.net.http.HttpClient;
import java.net.http.HttpRequest;
import java.net.http.HttpResponse;

// GET 请求示例
HttpClient client = HttpClient.newHttpClient();
HttpRequest request = HttpRequest.newBuilder()
        .uri(URI.create("https://api.example.com/users?page=1&limit=20"))
        .GET()
        .build();
HttpResponse<String> response = client.send(request, HttpResponse.BodyHandlers.ofString());
System.out.println(response.body());

POST 请求

POST 请求用于向服务器提交数据，数据通常放在请求体中。

python
# POST 请求示例
payload = {'username': 'admin', 'password': '123456'}
response = requests.post('https://api.example.com/login', data=payload)
java
// POST 请求示例
HttpRequest request = HttpRequest.newBuilder()
        .uri(URI.create("https://api.example.com/login"))
        .header("Content-Type", "application/x-www-form-urlencoded")
        .POST(HttpRequest.BodyPublishers.ofString("username=admin&password=123456"))
        .build();

请求头 (Headers)

请求头是 HTTP 请求的元数据，对于爬虫来说，正确设置请求头是避免被识别为爬虫的关键。

请求头说明重要性
User-Agent标识客户端类型和版本高
Cookie携带会话信息高
Referer标识请求来源页面中
Accept-Language可接受的语言类型中
Accept可接受的响应内容类型低
Origin请求来源域名（CORS）低

User-Agent

User-Agent 标识发起请求的浏览器类型、版本和操作系统。服务器通过 User-Agent 判断请求是否来自真实的浏览器。

python
headers = {
    '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',
    'Accept-Language': 'zh-CN,zh;q=0.9,en;q=0.8',
}
response = requests.get('https://example.com', headers=headers)
java
HttpRequest request = HttpRequest.newBuilder()
        .uri(URI.create("https://example.com"))
        .header("User-Agent", "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36")
        .header("Accept-Language", "zh-CN,zh;q=0.9")
        .GET()
        .build();

Cookie 与 Session

Cookie 和 Session 用于维持 HTTP 无状态协议下的用户状态。

特性CookieSession
存储位置客户端（浏览器）服务器端
存储容量约 4KB理论上较大
安全性较低，可被篡改较高
生命周期可设置过期时间通常会话结束即销毁
性能影响无服务器需维护状态

Session 通常通过 Cookie 中的 Session ID 来关联，服务器根据 Session ID 查找对应的 Session 数据。

python
# Session 会话保持
session = requests.Session()
session.cookies.set('session_id', 'abc123')
# 后续请求自动携带 Cookie
response = session.get('https://example.com/profile')
java
import java.net.CookieManager;
import java.net.CookieStore;

// Cookie 持久化
CookieManager cookieManager = new CookieManager();
HttpClient client = HttpClient.newBuilder()
        .cookieHandler(cookieManager)
        .build();
// 后续请求自动管理 Cookie

Referer 来源

Referer 请求头指示了当前请求的来源页面地址。许多网站会校验 Referer 防止跨站请求。

python
headers = {
    'Referer': 'https://example.com/previous-page',
    'User-Agent': 'Mozilla/5.0 ...',
}
requests.get('https://example.com/target', headers=headers)

HTML 解析 ​

爬虫获取到 HTML 后，需要从中提取目标数据。常见的解析方式包括 DOM 树操作、CSS Selector 和 XPath。

DOM 树

HTML 文档被解析为树状结构的 DOM（Document Object Model），每个 HTML 标签对应一个 DOM 节点，节点之间存在父子、兄弟关系。

html
<html>
  <body>
    <div class="container">
      <ul id="list">
        <li class="item">Item 1</li>
        <li class="item">Item 2</li>
      </ul>
    </div>
  </body>
</html>

CSS Selector 与 XPath 对比表

特性CSS SelectorXPath
语法简洁性简洁直观较复杂
功能广度选择元素选择元素 + 导航 + 计算
文本提取不支持直接提取文本支持 text() 函数
属性匹配[attr="val"][@attr="val"]
索引访问:nth-child(n)[n]
轴导航不支持支持 parent::、ancestor:: 等
解析速度较快较慢（功能更强大）
学习曲线低中
向后查找支持支持
向前查找不支持支持 preceding-sibling::

CSS Selector 示例

python
# Python (BeautifulSoup)
soup.select('div.container ul#list li.item')
soup.select_one('li.item')
python
# Python (lxml)
from lxml import etree
html = etree.HTML(content)
html.cssselect('div.container ul#list li.item')
java
// Java (Jsoup)
Document doc = Jsoup.parse(html);
Elements items = doc.select("div.container ul#list li.item");
Element firstItem = doc.selectFirst("li.item");

XPath 示例

python
from lxml import etree

html = etree.HTML(content)
# 绝对路径
items = html.xpath('/html/body/div/ul/li')
# 相对路径 + 属性过滤
items = html.xpath('//div[@class="container"]//li[@class="item"]')
# 文本提取
texts = html.xpath('//li[@class="item"]/text()')
# 索引
second = html.xpath('//li[@class="item"][2]')
java
// Java (使用 XPath 解析)
import javax.xml.parsers.DocumentBuilder;
import javax.xml.xpath.XPath;
import javax.xml.xpath.XPathFactory;

// 需配合 Jsoup 转换为 w3c Document
DocumentBuilder builder = DocumentBuilderFactory.newInstance().newDocumentBuilder();
InputSource source = new InputSource(new StringReader(html));
Document doc = builder.parse(source);
XPath xpath = XPathFactory.newInstance().newXPath();
NodeList nodes = (NodeList) xpath.evaluate("//div[@class='container']//li", doc, XPathConstants.NODESET);

Robots 协议 ​

Robots 协议（Robots Exclusion Protocol）通过 robots.txt 文件告知爬虫哪些路径可以访问、哪些路径禁止访问。该协议是行业规范，不具有法律约束力，但遵守它是有良好职业操守的表现。

robots.txt 解析

robots.txt 文件位于网站的根目录，例如 https://example.com/robots.txt。

User-agent: *
Disallow: /admin/
Disallow: /private/
Allow: /public/
Crawl-delay: 10

User-agent: Googlebot
Disallow: /temp/
指令说明
User-agent指定规则适用的爬虫名称，* 表示所有爬虫
Disallow禁止访问的路径，空值表示允许所有
Allow允许访问的路径（在 Disallow 限制下的例外）
Crawl-delay爬取间隔时间（秒）
Sitemap网站地图 URL

Python 解析 robots.txt

python
from urllib.robotparser import RobotFileParser

rp = RobotFileParser()
rp.set_url('https://example.com/robots.txt')
rp.read()

# 检测是否允许爬取
can_fetch = rp.can_fetch('MySpider/1.0', 'https://example.com/admin/')
print(f"允许爬取: {can_fetch}")

crawl_delay = rp.crawl_delay('MySpider/1.0')
print(f"爬取间隔: {crawl_delay}秒")

使用 Scrapy 的 Robots 协议支持

python
# settings.py
ROBOTSTXT_OBEY = True
USER_AGENT = 'MySpider (+http://example.com/bot)'

反爬与应对策略 ​

User-Agent 伪装 ​

网站通常会检查 User-Agent，拒绝非浏览器 User-Agent 的请求。应对策略是维护一个 User-Agent 池，随机切换。

python
import random

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/605.1.15 ... Safari/605.1.15',
    'Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:109.0) Gecko/20100101 Firefox/121.0',
    'Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 ... Chrome/119.0.0.0 Safari/537.36',
]

def get_random_headers():
    return {'User-Agent': random.choice(USER_AGENTS)}
java
import java.util.Arrays;
import java.util.List;
import java.util.Random;

public class UserAgentRotator {
    private static final List<String> USER_AGENTS = Arrays.asList(
        "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/605.1.15 ... Safari/605.1.15"
    );
    private static final Random RANDOM = new Random();

    public static String getRandomUserAgent() {
        return USER_AGENTS.get(RANDOM.nextInt(USER_AGENTS.size()));
    }
}

IP 代理池 ​

使用代理 IP 可以绕过基于 IP 的访问频率限制和封禁。

代理类型

协议类型说明适用场景
HTTP仅支持 HTTP 请求普通 HTTP 网站
HTTPS支持 HTTP 和 HTTPS 请求HTTPS 网站
SOCKS4支持 TCP 连接非 HTTP 协议
SOCKS5支持 TCP/UDP，支持认证通用场景

代理验证

代理 IP 在使用前需要验证其可用性、匿名性和响应速度。

python
import requests
from concurrent.futures import ThreadPoolExecutor, as_completed

def validate_proxy(proxy):
    """验证代理可用性"""
    proxies = {
        'http': f'http://{proxy}',
        'https': f'http://{proxy}',
    }
    try:
        start = time.time()
        response = requests.get('https://httpbin.org/ip', proxies=proxies, timeout=5)
        elapsed = time.time() - start
        if response.status_code == 200:
            return {'proxy': proxy, 'speed': elapsed, 'valid': True}
    except:
        pass
    return {'proxy': proxy, 'valid': False}

# 并发验证代理池
with ThreadPoolExecutor(max_workers=20) as executor:
    futures = [executor.submit(validate_proxy, p) for p in proxy_list]
    for future in as_completed(futures):
        result = future.result()
        if result['valid']:
            valid_proxies.append(result)
java
import java.net.URI;
import java.net.http.HttpClient;
import java.net.http.HttpRequest;
import java.net.http.HttpResponse;
import java.net.InetSocketAddress;
import java.net.ProxySelector;

// 使用代理
ProxySelector proxySelector = ProxySelector.of(new InetSocketAddress("127.0.0.1", 8080));
HttpClient client = HttpClient.newBuilder()
        .proxy(proxySelector)
        .connectTimeout(Duration.ofSeconds(5))
        .build();

轮换策略

python
import itertools
import time

class ProxyPool:
    def __init__(self, proxies, max_fails=3, ban_cooldown=60):
        self.proxies = list(proxies)
        self.fail_count = {p: 0 for p in self.proxies}
        self.banned_until = {}
        self.max_fails = max_fails
        self.ban_cooldown = ban_cooldown
        self.cycle = itertools.cycle(self.proxies)

    def get_proxy(self):
        for _ in range(len(self.proxies)):
            proxy = next(self.cycle)
            if proxy in self.banned_until:
                if time.time() < self.banned_until[proxy]:
                    continue
                else:
                    del self.banned_until[proxy]
                    self.fail_count[proxy] = 0
            return proxy
        return None

    def report_failure(self, proxy):
        self.fail_count[proxy] = self.fail_count.get(proxy, 0) + 1
        if self.fail_count[proxy] >= self.max_fails:
            self.banned_until[proxy] = time.time() + self.ban_cooldown

请求量控制

控制请求频率避免触发反爬机制。

python
import time
import threading

class RateLimiter:
    def __init__(self, calls_per_second=1):
        self.min_interval = 1.0 / calls_per_second
        self.last_call = 0
        self.lock = threading.Lock()

    def wait(self):
        with self.lock:
            now = time.time()
            elapsed = now - self.last_call
            if elapsed < self.min_interval:
                time.sleep(self.min_interval - elapsed)
            self.last_call = time.time()

请求头伪装 ​

现代网站（尤其是使用 CloudFlare 等 CDN 的网站）会检查请求头链的完整性。完整的请求头伪装应包括以下字段：

python
HEADERS_TEMPLATE = {
    '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',
    'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8',
    'Accept-Language': 'zh-CN,zh;q=0.9,en;q=0.8',
    'Accept-Encoding': 'gzip, deflate, br',
    'Referer': 'https://www.google.com/',
    'Origin': 'https://example.com',
    'Connection': 'keep-alive',
    'Upgrade-Insecure-Requests': '1',
    'Sec-Fetch-Site': 'same-origin',
    'Sec-Fetch-Mode': 'navigate',
    'Sec-Fetch-Dest': 'document',
    'Sec-Fetch-User': '?1',
    'Sec-Ch-Ua': '"Not_A Brand";v="8", "Chromium";v="120", "Google Chrome";v="120"',
    'Sec-Ch-Ua-Mobile': '?0',
    'Sec-Ch-Ua-Platform': '"Windows"',
    'DNT': '1',
}
请求头含义典型值
Accept可接受的响应类型text/html,application/xhtml+xml,...
Accept-Language可接受的语言zh-CN,zh;q=0.9,en;q=0.8
Accept-Encoding可接受的编码gzip, deflate, br
Referer请求来源前一个页面的 URL
Origin请求来源域名https://example.com
Sec-Fetch-Site请求发起者与目标的关系same-origin, cross-site, same-site
Sec-Fetch-Mode请求模式navigate, cors, no-cors, same-origin
Sec-Fetch-Dest请求目标类型document, image, script, empty
Sec-Fetch-User是否由用户操作触发?1
Sec-Ch-Ua客户端品牌信息Chromium 版本信息
Sec-Ch-Ua-Platform操作系统平台"Windows", "macOS"

验证码识别 ​

验证码（CAPTCHA）是常见的反爬手段。应对策略根据验证码类型和复杂度不同而异。

验证码类型与应对方案

验证码类型应对方案识别率成本
简单数字/字母验证码OCR Tesseract60%-80%免费
扭曲字符验证码OCR + 预处理70%-90%免费
中文验证码第三方打码平台80%-95%低
滑块验证码第三方打码平台 / Selenium 模拟70%-95%中
点选验证码第三方打码平台80%-90%中
reCAPTCHA v2/v3第三方打码平台 / Capsolver 服务60%-90%高
自定义复杂验证码机器学习 + CNN90%+高

OCR Tesseract 示例

python
import pytesseract
from PIL import Image, ImageFilter, ImageEnhance

def preprocess_image(image_path):
    """验证码图片预处理"""
    img = Image.open(image_path)
    # 灰度化
    img = img.convert('L')
    # 增强对比度
    enhancer = ImageEnhance.Contrast(img)
    img = enhancer.enhance(2)
    # 二值化
    threshold = 128
    img = img.point(lambda x: 0 if x < threshold else 255)
    # 降噪
    img = img.filter(ImageFilter.MedianFilter())
    return img

def recognize_captcha(image_path):
    """OCR 识别验证码"""
    img = preprocess_image(image_path)
    config = '--psm 8 -c tessedit_char_whitelist=0123456789ABCDEFGHIJKLMNOPQRSTUVWXYZ'
    text = pytesseract.image_to_string(img, config=config)
    return text.strip()

第三方打码平台

python
class CaptchaSolver:
    """第三方打码平台封装（示例对接 2captcha）"""
    def __init__(self, api_key):
        self.api_key = api_key
        self.base_url = 'https://2captcha.com'

    def solve_normal(self, image_base64):
        """识别普通验证码"""
        payload = {
            'key': self.api_key,
            'method': 'base64',
            'body': image_base64,
        }
        resp = requests.post(f'{self.base_url}/in.php', data=payload)
        captcha_id = resp.text.split('|')[1]

        # 轮询获取结果
        for _ in range(30):
            result = requests.get(
                f'{self.base_url}/res.php',
                params={'key': self.api_key, 'action': 'get', 'id': captcha_id}
            )
            if result.text == 'CAPCHA_NOT_READY':
                time.sleep(2)
                continue
            return result.text.split('|')[1]
        return None

    def solve_recaptcha(self, site_key, page_url):
        """识别 reCAPTCHA"""
        payload = {
            'key': self.api_key,
            'method': 'userrecaptcha',
            'googlekey': site_key,
            'pageurl': page_url,
        }
        resp = requests.post(f'{self.base_url}/in.php', data=payload)
        captcha_id = resp.text.split('|')[1]

        for _ in range(60):
            result = requests.get(
                f'{self.base_url}/res.php',
                params={'key': self.api_key, 'action': 'get', 'id': captcha_id}
            )
            if 'OK' in result.text:
                return result.text.split('|')[1]
            time.sleep(5)
        return None

动态渲染 ​

对于 JavaScript 动态渲染的页面（SPA 应用、Ajax 加载内容等），需要使用无头浏览器或渲染服务获取完整的 HTML。

工具对比表

特性SeleniumPlaywrightSplashPuppeteer
语言支持Python/Java/JS 等Python/JS/Java/.NETLua 脚本JavaScript
浏览器引擎Chrome/Firefox/EdgeChromium/Firefox/WebKitWebKitChromium
执行速度慢快中快
资源占用高中中中
并发支持需手动管理原生支持 Browser Context通过 API 调用需手动管理
网络控制有限强大（路由拦截）基本强大
截图/PDF支持支持支持支持
维护状态稳定维护活跃开发维护模式稳定维护
学习曲线低中中中

渲染流程区别

渲染方式流程适用场景
服务端渲染 (SSR)服务器返回完整 HTML可直接获取内容，无需渲染
客户端渲染 (CSR/SPA)服务器返回空壳 HTML + JS，浏览器执行 JS 渲染需无头浏览器
静态站点生成 (SSG)构建时生成完整 HTML内容已固化，直接抓取即可

Selenium 示例

python
from selenium import webdriver
from selenium.webdriver.chrome.options import Options
from selenium.webdriver.common.by import By
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC

options = Options()
options.add_argument('--headless')  # 无头模式
options.add_argument('--no-sandbox')
options.add_argument('--disable-dev-shm-usage')
options.add_argument('user-agent=Mozilla/5.0 (Windows NT 10.0; Win64; x64) ...')

driver = webdriver.Chrome(options=options)
driver.get('https://example.com')

# 等待动态内容加载
wait = WebDriverWait(driver, 10)
element = wait.until(EC.presence_of_element_located((By.CLASS_NAME, 'dynamic-content')))

# 获取渲染后的 HTML
html = driver.page_source
driver.quit()
java
import org.openqa.selenium.WebDriver;
import org.openqa.selenium.chrome.ChromeDriver;
import org.openqa.selenium.chrome.ChromeOptions;
import org.openqa.selenium.support.ui.WebDriverWait;
import org.openqa.selenium.support.ui.ExpectedConditions;
import org.openqa.selenium.By;

ChromeOptions options = new ChromeOptions();
options.addArguments("--headless");
options.addArguments("user-agent=Mozilla/5.0 ...");

WebDriver driver = new ChromeDriver(options);
driver.get("https://example.com");

WebDriverWait wait = new WebDriverWait(driver, Duration.ofSeconds(10));
wait.until(ExpectedConditions.presenceOfElementLocated(By.className("dynamic-content")));

String html = driver.getPageSource();
driver.quit();

Playwright 示例

python
from playwright.sync_api import sync_playwright

with sync_playwright() as p:
    browser = p.chromium.launch(headless=True)
    context = browser.new_context(
        user_agent='Mozilla/5.0 (Windows NT 10.0; Win64; x64) ...',
        viewport={'width': 1920, 'height': 1080},
    )
    page = context.new_page()
    page.goto('https://example.com')

    # 等待元素出现
    page.wait_for_selector('.dynamic-content', timeout=10000)
    # 等待网络请求完成
    page.wait_for_load_state('networkidle')

    # 获取渲染后 HTML
    html = page.content()

    # 拦截网络请求（高级用法）
    page.route('**/api/**', lambda route: route.continue_())

    browser.close()

性能对比

场景Selenium (Chrome)Playwright (Chromium)Splash
首次启动时间2-3s1-2s0.5-1s
单页面渲染1-3s0.5-2s0.5-1.5s
内存占用200-400MB100-250MB50-150MB
并发 10 个实例高中中

Python 爬虫 ​

Requests 库 ​

Requests 是 Python 最流行的 HTTP 库，提供了简洁的 API。

基本请求

python
import requests

# GET 请求
resp = requests.get('https://api.example.com/data', params={'id': 1})

# POST 请求
resp = requests.post('https://api.example.com/submit', json={'name': 'test'})

# PUT / DELETE / PATCH
requests.put('https://api.example.com/update', data={'key': 'value'})
requests.delete('https://api.example.com/delete/1')
requests.patch('https://api.example.com/patch', json={'field': 'new_value'})

Session 与会话保持

python
session = requests.Session()
session.headers.update({'User-Agent': 'Mozilla/5.0 ...'})

# 登录（Cookie 自动保存到 Session）
login_resp = session.post('https://example.com/login', data={
    'username': 'user',
    'password': 'pass',
})

# 后续请求自动携带 Cookie
profile_resp = session.get('https://example.com/profile')

超时与重试

python
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

session = requests.Session()

# 配置重试策略
retry_strategy = Retry(
    total=3,                    # 总重试次数
    backoff_factor=1,           # 退避因子（1s, 2s, 4s）
    status_forcelist=[500, 502, 503, 504],  # 重试的 HTTP 状态码
    allowed_methods=['GET', 'POST'],
)

adapter = HTTPAdapter(max_retries=retry_strategy)
session.mount('https://', adapter)
session.mount('http://', adapter)

# 超时（连接超时 + 读取超时）
response = session.get('https://example.com', timeout=(3, 10))

代理设置

python
proxies = {
    'http': 'http://user:pass@127.0.0.1:8080',
    'https': 'http://user:pass@127.0.0.1:8080',
    # SOCKS5 代理（需要安装 requests[socks]）
    # 'http': 'socks5://127.0.0.1:1080',
    # 'https': 'socks5://127.0.0.1:1080',
}
response = requests.get('https://example.com', proxies=proxies)

SSL 验证

python
# 禁用 SSL 验证（不推荐生产环境）
response = requests.get('https://example.com', verify=False)

# 使用自定义 CA 证书
response = requests.get('https://example.com', verify='/path/to/cert.pem')

# 使用客户端证书
response = requests.get('https://example.com', cert=('/path/to/client.cert', '/path/to/client.key'))

Cookie 持久化

python
import json

def save_cookies(session, filepath):
    """保存 Cookie 到文件"""
    cookies = session.cookies.get_dict()
    with open(filepath, 'w') as f:
        json.dump(cookies, f)

def load_cookies(session, filepath):
    """从文件加载 Cookie"""
    with open(filepath, 'r') as f:
        cookies = json.load(f)
    session.cookies.update(cookies)
    return session

BeautifulSoup ​

BeautifulSoup 是 Python 最流行的 HTML/XML 解析库。

解析器对比

解析器类型速度容错性依赖
html.parserPython 标准库慢中无需安装
lxmlC 扩展快高需安装 lxml
html5libPython 纯实现最慢最高（浏览器级）需安装 html5lib
xml (lxml-xml)C 扩展快中需安装 lxml

Parser 差异示例

python
from bs4 import BeautifulSoup

html = '<html><br/><p>text</p></html>'

# html.parser - 标准库解析器
soup1 = BeautifulSoup(html, 'html.parser')
# 可能不会自动闭合某些标签

# lxml - 快速且容错
soup2 = BeautifulSoup(html, 'lxml')
# 自动补充缺失的 html/body 标签

# html5lib - 浏览器级容错
soup3 = BeautifulSoup(html, 'html5lib')
# 最大程度模拟浏览器行为，结果最规范

find / find_all

python
soup = BeautifulSoup(html, 'lxml')

# find_all - 查找所有匹配元素
all_divs = soup.find_all('div')
all_items = soup.find_all(class_='item')           # 按 class
all_active = soup.find_all('li', class_='active')  # 标签 + class
all_by_attrs = soup.find_all(attrs={'data-id': '123'})

# find - 查找第一个匹配元素
first_div = soup.find('div')
first_link = soup.find('a', href=True)  # 存在 href 属性的 a 标签

# 使用正则表达式
import re
emails = soup.find_all(text=re.compile(r'[\w.+-]+@[\w-]+\.[\w.-]+'))

# 限制返回数量
first_three = soup.find_all('div', limit=3)

# 递归控制
direct_children = soup.find_all('div', recursive=False)

CSS Selector

python
# select - 返回列表
items = soup.select('div.container > ul#list > li.item')
first = soup.select_one('div.container > ul#list > li.item')

# 属性选择器
soup.select('a[href^="https"]')           # href 以 https 开头
soup.select('a[href$=".pdf"]')            # href 以 .pdf 结尾
soup.select('a[href*="download"]')        # href 包含 download
soup.select('[data-type="article"]')      # 自定义属性

# 伪类选择器
soup.select('li:nth-child(2)')            # 第二个 li
soup.select('li:first-child')             # 第一个 li
soup.select('li:last-child')              # 最后一个 li
soup.select('p:empty')                    # 空 p 标签
soup.select(':not(.hidden)')              # 不包含 hidden 类的元素

Tag 属性与 Navigation

python
# tag 属性
tag = soup.find('a')
print(tag.name)           # 标签名: 'a'
print(tag.attrs)          # 所有属性: {'href': '...', 'class': ['link']}
print(tag['href'])        # 获取属性值
print(tag.get('href'))    # 安全获取属性值
print(tag.get('target', '_self'))  # 带默认值

# tag 内容
print(tag.string)         # 标签内文本（不含子标签）
print(tag.text)           # 所有文本（含子标签）
print(tag.get_text(separator=' ', strip=True))

# Navigation 遍历
tag.parent                # 父节点
tag.parents               # 所有祖先节点（生成器）
tag.children              # 直接子节点（生成器）
tag.descendants           # 所有子孙节点（生成器）
tag.next_sibling          # 下一个兄弟节点
tag.previous_sibling      # 上一个兄弟节点
tag.next_siblings         # 之后所有兄弟节点（生成器）
tag.previous_siblings     # 之前所有兄弟节点（生成器）

# 查找兄弟节点
tag.find_next_sibling('div')
tag.find_previous_sibling('div')
tag.find_all_next('p')
tag.find_all_previous('p')

Scrapy 框架 ​

Scrapy 是 Python 最强大的爬虫框架，提供了完整的爬取、处理、存储流水线。

项目结构

myproject/
├── scrapy.cfg                  # 项目配置
├── myproject/
│   ├── __init__.py
│   ├── items.py                # Item 定义（数据模型）
│   ├── middlewares.py           # 中间件
│   ├── pipelines.py            # Pipeline（数据处理与存储）
│   ├── settings.py             # 全局设置
│   ├── spiders/                # Spider 目录
│   │   ├── __init__.py
│   │   └── example_spider.py
│   └── itemloaders.py          # Item Loader（可选）

Spider 示例

python
import scrapy
from myproject.items import ProductItem

class ProductSpider(scrapy.Spider):
    name = 'product_spider'
    allowed_domains = ['example.com']
    start_urls = ['https://example.com/products']

    def parse(self, response):
        """解析列表页"""
        for product in response.css('div.product-item'):
            item = ProductItem()
            item['name'] = product.css('h2::text').get()
            item['price'] = product.css('span.price::text').get()
            item['url'] = product.css('a::attr(href)').get()

            # 跟进详情页
            yield response.follow(item['url'], self.parse_detail, meta={'item': item})

        # 分页
        next_page = response.css('a.next::attr(href)').get()
        if next_page:
            yield response.follow(next_page, self.parse)

    def parse_detail(self, response):
        """解析详情页"""
        item = response.meta['item']
        item['description'] = response.css('div.description::text').get()
        item['specs'] = response.css('table.specs td::text').getall()
        yield item

Item 定义

python
# items.py
import scrapy

class ProductItem(scrapy.Item):
    name = scrapy.Field()
    price = scrapy.Field()
    url = scrapy.Field()
    description = scrapy.Field()
    specs = scrapy.Field()
    crawl_time = scrapy.Field()

Pipeline 数据处理

python
# pipelines.py
import json
import pymongo
from itemadapter import ItemAdapter

class JsonPipeline:
    """JSON 文件存储"""
    def open_spider(self, spider):
        self.file = open('products.json', 'w', encoding='utf-8')
        self.file.write('[')

    def close_spider(self, spider):
        self.file.write(']')
        self.file.close()

    def process_item(self, item, spider):
        line = json.dumps(ItemAdapter(item).asdict(), ensure_ascii=False) + ',\n'
        self.file.write(line)
        return item

class MongoPipeline:
    """MongoDB 存储"""
    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['products'].insert_one(ItemAdapter(item).asdict())
        return item

class DuplicatesPipeline:
    """去重 Pipeline"""
    def __init__(self):
        self.urls_seen = set()

    def process_item(self, item, spider):
        adapter = ItemAdapter(item)
        if adapter.get('url') in self.urls_seen:
            raise DropItem(f"重复项: {adapter['url']}")
        self.urls_seen.add(adapter['url'])
        return item

Middleware 中间件

python
# middlewares.py
import random
from scrapy import signals
from scrapy.downloadermiddlewares.useragent import UserAgentMiddleware

class RandomUserAgentMiddleware(UserAgentMiddleware):
    """随机 User-Agent"""
    def __init__(self, user_agents):
        self.user_agents = user_agents

    @classmethod
    def from_crawler(cls, crawler):
        return cls(
            user_agents=crawler.settings.get('USER_AGENT_LIST', [])
        )

    def process_request(self, request, spider):
        request.headers['User-Agent'] = random.choice(self.user_agents)

class ProxyMiddleware:
    """代理中间件"""
    def process_request(self, request, spider):
        request.meta['proxy'] = 'http://127.0.0.1:8080'

class RetryMiddleware:
    """自定义重试中间件"""
    def process_response(self, request, response, spider):
        if response.status in [403, 429]:
            # 被封 IP 时的处理
            spider.logger.warning(f"请求被限制: {request.url}")
            # 切换代理重试
            new_request = request.copy()
            new_request.meta['proxy'] = get_new_proxy()
            return new_request
        return response

Settings 配置

python
# settings.py
BOT_NAME = 'myproject'
SPIDER_MODULES = ['myproject.spiders']
NEWSPIDER_MODULE = 'myproject.spiders'

USER_AGENT = 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) ...'
ROBOTSTXT_OBEY = True

# 下载延迟
DOWNLOAD_DELAY = 1.0        # 基础延迟（秒）
RANDOMIZE_DOWNLOAD_DELAY = True  # 随机化延迟（0.5x ~ 1.5x）

# 并发控制
CONCURRENT_REQUESTS = 16           # 全局并发请求数
CONCURRENT_REQUESTS_PER_DOMAIN = 8 # 单域名并发
CONCURRENT_REQUESTS_PER_IP = 0     # 单 IP 并发（0 表示不限制）
REACTOR_THREADPOOL_MAXSIZE = 20    # 线程池大小

# 重试设置
RETRY_ENABLED = True
RETRY_TIMES = 3
RETRY_HTTP_CODES = [500, 502, 503, 504, 408, 429]

# 超时设置
DOWNLOAD_TIMEOUT = 30
CONNECT_TIMEOUT = 10

# Pipeline 配置（数值越小优先级越高）
ITEM_PIPELINES = {
    'myproject.pipelines.DuplicatesPipeline': 100,
    'myproject.pipelines.JsonPipeline': 200,
    'myproject.pipelines.MongoPipeline': 300,
}

# Middleware 配置
DOWNLOADER_MIDDLEWARES = {
    'myproject.middlewares.RandomUserAgentMiddleware': 400,
    'myproject.middlewares.ProxyMiddleware': 500,
    'scrapy.downloadermiddlewares.retry.RetryMiddleware': 550,
}

# 自定义设置
MONGO_URI = 'mongodb://localhost:27017'
MONGO_DATABASE = 'scrapy_data'
USER_AGENT_LIST = [
    'Mozilla/5.0 (Windows NT 10.0; Win64; x64) ...',
    'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) ...',
    'Mozilla/5.0 (X11; Linux x86_64) ...',
]

Scrapy 进阶 ​

CrawlSpider

CrawlSpider 通过 Rule 规则自动跟进链接，适合爬取整站内容。

python
from scrapy.spiders import CrawlSpider, Rule
from scrapy.linkextractors import LinkExtractor

class BlogCrawlSpider(CrawlSpider):
    name = 'blog_crawler'
    allowed_domains = ['blog.example.com']
    start_urls = ['https://blog.example.com/']

    rules = (
        # 提取分类页链接
        Rule(LinkExtractor(allow=r'/category/\w+/$'), follow=True),

        # 提取分页链接
        Rule(LinkExtractor(allow=r'/page/\d+/$'), follow=True),

        # 提取文章详情链接（最终爬取目标）
        Rule(LinkExtractor(allow=r'/article/\d+/$'), callback='parse_article'),
    )

    def parse_article(self, response):
        yield {
            'title': response.css('h1.title::text').get(),
            'content': response.css('div.content').get(),
            'url': response.url,
        }

Item Loader

Item Loader 提供了一种声明式的数据清洗和处理方式。

python
from scrapy.loader import ItemLoader
from scrapy.loader.processors import TakeFirst, MapCompose, Join
from itemloaders.processors import Identity
import re

def strip_whitespace(value):
    return value.strip() if value else value

def parse_price(value):
    """解析价格（去除货币符号）"""
    if value:
        return float(re.sub(r'[^\d.]+', '', value))
    return 0.0

def parse_stock(value):
    return value.strip() == '有货'

class ProductLoader(ItemLoader):
    default_output_processor = TakeFirst()

    name_in = MapCompose(str.strip)
    price_in = MapCompose(parse_price)
    description_in = MapCompose(strip_whitespace)
    in_stock_in = MapCompose(parse_stock)
    images_out = Identity()  # 保持为列表
    tags_out = Identity()

# 使用 Item Loader
loader = ProductLoader(item=ProductItem(), response=response)
loader.add_css('name', 'h1.product-title::text')
loader.add_css('price', 'span.price::text')
loader.add_css('description', 'div.description::text')
loader.add_css('in_stock', 'span.stock-status::text')
loader.add_css('images', 'img.product-image::attr(src)')
loader.add_value('url', response.url)
yield loader.load_item()

ImagesPipeline / FilesPipeline

Scrapy 内置了媒体文件下载 Pipeline。

python
# settings.py
ITEM_PIPELINES = {
    'scrapy.pipelines.images.ImagesPipeline': 1,
    'scrapy.pipelines.files.FilesPipeline': 2,
}

IMAGES_STORE = '/data/images'       # 图片存储目录
FILES_STORE = '/data/files'         # 文件存储目录

IMAGES_THUMBS = {                   # 自动生成缩略图
    'small': (50, 50),
    'medium': (200, 200),
}

IMAGES_MIN_HEIGHT = 100             # 最小高度过滤
IMAGES_MIN_WIDTH = 100              # 最小宽度过滤
IMAGES_EXPIRES = 90                 # 缓存过期天数
FILES_EXPIRES = 90
python
# items.py - 图片和文件 Item
class ProductWithImagesItem(scrapy.Item):
    name = scrapy.Field()
    image_urls = scrapy.Field()       # 必须命名为 image_urls（ImagesPipeline）
    images = scrapy.Field()           # 下载结果自动写入
    file_urls = scrapy.Field()        # 必须命名为 file_urls（FilesPipeline）
    files = scrapy.Field()            # 下载结果自动写入

# 自定义 ImagesPipeline
from scrapy.pipelines.images import ImagesPipeline
from scrapy.http import Request

class CustomImagesPipeline(ImagesPipeline):
    def file_path(self, request, response=None, info=None, *, item=None):
        """自定义文件存储路径"""
        image_name = request.url.split('/')[-1]
        return f'full/{item["name"]}/{image_name}'

    def item_completed(self, results, item, info):
        if results:
            item['images'] = [result['path'] for ok, result in results if ok]
        return item

分布式爬虫 Scrapy-Redis

Scrapy-Redis 将 Scrapy 的调度器和去重集合迁移到 Redis，实现多台机器协同爬取。

python
# settings.py
# 启用 Scrapy-Redis 调度器
SCHEDULER = 'scrapy_redis.scheduler.Scheduler'
SCHEDULER_PERSIST = True  # 爬取暂停后不丢失队列

# 使用 Redis 去重
DUPEFILTER_CLASS = 'scrapy_redis.dupefilter.RFPDupeFilter'

# Redis 连接配置
REDIS_URL = 'redis://localhost:6379'

# 可选的：从 Redis 队列读取 start_urls
REDIS_START_URLS_KEY = '%(name)s:start_urls'
python
# spider.py - 基于 Scrapy-Redis 的 Spider
from scrapy_redis.spiders import RedisSpider

class RedisProductSpider(RedisSpider):
    """从 Redis 队列读取起始 URL"""
    name = 'redis_product_spider'
    allowed_domains = ['example.com']

    # 从 Redis key 'redis_product_spider:start_urls' 读取起始 URL
    redis_key = '%(name)s:start_urls'

    def parse(self, response):
        # 同普通 Spider 的 parse 方法
        pass

# 推送起始 URL 到 Redis:
# redis-cli lpush redis_product_spider:start_urls https://example.com/products

# 启动多个爬虫实例（多台机器）：
# scrapy crawl redis_product_spider

Java 爬虫 ​

Jsoup ​

Jsoup 是 Java 平台最流行的 HTML 解析库，提供了 jQuery 风格的 API。

parse 解析

java
import org.jsoup.Jsoup;
import org.jsoup.nodes.Document;
import org.jsoup.nodes.Element;
import org.jsoup.select.Elements;
import java.io.File;
import java.io.IOException;

// 解析 HTML 字符串
String html = "<html><body><div class='content'>Hello</div></body></html>";
Document doc = Jsoup.parse(html);

// 解析 URL（直接获取并解析）
Document docFromUrl = Jsoup.connect("https://example.com")
        .userAgent("Mozilla/5.0 ...")
        .timeout(10000)
        .get();

// 解析文件
Document docFromFile = Jsoup.parse(new File("index.html"), "UTF-8");

// 解析片段
Document fragment = Jsoup.parseBodyFragment("<div><p>片段</p></div>");

Connection 设置

java
// 完整 Connection 配置
Document doc = Jsoup.connect("https://example.com/login")
        .url("https://example.com/login")
        .userAgent("Mozilla/5.0 ...")
        .timeout(10000)
        .cookie("session_id", "abc123")
        .referrer("https://google.com")
        .header("Accept-Language", "zh-CN,zh;q=0.9")
        .method(Connection.Method.POST)
        .data("username", "admin")
        .data("password", "123456")
        .followRedirects(true)
        .ignoreHttpErrors(false)
        .maxBodySize(0) // 无限制
        .execute()
        .parse();

Selector 选择器

java
Document doc = Jsoup.connect("https://example.com").get();

// CSS Selector
Elements links = doc.select("a[href]");                    // 所有带 href 的 a 标签
Elements images = doc.select("img[src$=.jpg]");           // 所有 .jpg 图片
Elements items = doc.select("div.item, li.product");      // 多选择器
Element first = doc.selectFirst("h1.title");              // 第一个匹配

// 属性选择
Elements externalLinks = doc.select("a[href^=https]");    // href 以 https 开头
Elements pdfLinks = doc.select("a[href$=.pdf]");          // href 以 .pdf 结尾
Elements containsLinks = doc.select("a[href*=download]"); // href 包含 download

// 伪类选择器
Elements oddRows = doc.select("tr:odd");                  // 奇数行
Elements evenRows = doc.select("tr:even");                // 偶数行
Element lastItem = doc.select("li:last-child").first();   // 最后一个 li

// 父子/兄弟选择器
Elements directChildren = doc.select("ul > li");          // 直接子元素
Elements descendants = doc.select("ul li");               // 所有后代
Elements siblings = doc.select("li + li");                // 相邻兄弟

Elements / Element 操作

java
// Elements 集合操作
Elements items = doc.select("li.product");
System.out.println(items.size());          // 数量
items.first();                             // 第一个
items.last();                              // 最后一个
items.get(2);                              // 索引（从 0 开始）
items.subList(0, 5);                       // 子列表

// 遍历
for (Element item : items) {
    // 处理每个元素
}

// Element 操作
Element div = doc.selectFirst("div.content");
div.tagName();                             // 标签名
div.id();                                  // id 属性
div.className();                           // class 属性
div.classNames();                          // 所有 class（Set<String>）
div.hasClass("active");                    // 是否包含 class
div.attributes();                          // 所有属性
div.dataset();                             // data-* 属性（Map<String, String>）

// DOM 遍历
div.parent();                              // 父节点
div.parents();                             // 所有祖先
div.children();                            // 直接子元素
div.childElements();                       // 直接子元素（Element 版本）
div.firstElementChild();                   // 第一个子元素
div.lastElementChild();                    // 最后一个子元素
div.siblingElements();                     // 兄弟元素
div.nextElementSibling();                  // 下一个兄弟元素
div.previousElementSibling();              // 上一个兄弟元素

attributes / text / html

java
// 属性操作
Element link = doc.selectFirst("a");
link.attr("href");                         // 获取 href 属性值
link.attr("target", "_blank");             // 设置属性
link.removeAttr("rel");                    // 移除属性
link.hasAttr("download");                  // 是否包含属性
link.absUrl("href");                       // 解析为绝对 URL

// 文本操作
Element p = doc.selectFirst("p.content");
p.text();                                  // 纯文本（不含标签）
p.wholeText();                             // 全部文本（含空白）
p.textNodes();                             // 所有 TextNode
p.getOwnText();                            // 仅自身的直接文本

// HTML 操作
div.html();                                // 内部 HTML
div.outerHtml();                           // 外部 HTML（含自身标签）
div.data();                                // 数据内容（script/style）

// 数据提取综合示例
Elements products = doc.select("div.product");
for (Element product : products) {
    String name = product.selectFirst("h2.name").text();
    double price = Double.parseDouble(
        product.selectFirst("span.price").text().replace("$", "")
    );
    String url = product.selectFirst("a").absUrl("href");
    String image = product.selectFirst("img").attr("src");
    System.out.printf("%s: $%.2f [%s]%n", name, price, url);
}

WebMagic ​

WebMagic 是 Java 生态中类似于 Scrapy 的爬虫框架，采用可扩展的组件架构。

Site 配置

java
import us.codecraft.webmagic.Site;

public Site createSite() {
    return Site.me()
            .setDomain("example.com")
            .setUserAgent("Mozilla/5.0 (Windows NT 10.0; Win64; x64) ...")
            .setCharset("UTF-8")
            .setTimeOut(10000)
            .setRetryTimes(3)
            .setSleepTime(1000)                      // 请求间隔（毫秒）
            .setCycleRetryTimes(3)                    // 循环重试次数
            .setRetrySleepTime(1000)                  // 重试等待时间
            .addHeader("Accept-Language", "zh-CN,zh;q=0.9")
            .addHeader("Referer", "https://google.com")
            .addCookie("session_id", "abc123")
            .setHttpProxy("127.0.0.1", 8080);         // HTTP 代理
}

PageProcessor

PageProcessor 是 WebMagic 的核心，负责页面解析和链接提取。

java
import us.codecraft.webmagic.Page;
import us.codecraft.webmagic.Site;
import us.codecraft.webmagic.processor.PageProcessor;
import us.codecraft.webmagic.selector.Selectable;

public class ProductPageProcessor implements PageProcessor {
    private Site site = Site.me()
            .setRetryTimes(3)
            .setSleepTime(1000)
            .setUserAgent("Mozilla/5.0 ...");

    @Override
    public void process(Page page) {
        // 提取数据
        page.putField("title", page.getHtml().xpath("//h1[@class='title']/text()"));
        page.putField("price", page.getHtml().css("span.price", "text"));
        page.putField("description", page.getHtml().css("div.description").get());
        page.putField("url", page.getUrl().get());

        // 提取链接并加入爬取队列
        page.addTargetRequests(
            page.getHtml().links().regex("https://example\\.com/product/\\d+").all()
        );

        // 分页
        page.addTargetRequest(
            page.getHtml().xpath("//a[@class='next']/@href").get()
        );
    }

    @Override
    public Site getSite() {
        return site;
    }
}

Selectable 选择器链

WebMagic 的 Selectable 支持链式调用，包括 XPath、CSS、Regex、Links 等多种选择器。

java
// Selectable 选择器链
Selectable html = page.getHtml();

// CSS 选择器
String text = html.css("div.content", "text").get();           // 获取文本
List<String> texts = html.css("li.item", "text").all();        // 获取所有文本
String htmlContent = html.css("div.content").get();            // 获取 HTML

// XPath 选择器
String title = html.xpath("//h1[@class='title']/text()").get();
String link = html.xpath("//a/@href").get();
String attr = html.xpath("//img/@src").get();

// 正则表达式
String regex = html.css("script").regex("window\\.data=(\\{.+?\\})").get();

// Links 提取
List<String> links = html.links().all();                         // 所有链接
List<String> productLinks = html.links()
        .regex("https://example\\.com/product/\\d+")
        .all();

// 链式组合
String result = page.getHtml()
        .css("div.product")
        .xpath("//h2")
        .css("a", "text")
        .get();

Pipeline

Pipeline 负责处理 PageProcessor 提取的结果（数据存储）。

java
import us.codecraft.webmagic.ResultItems;
import us.codecraft.webmagic.Task;
import us.codecraft.webmagic.pipeline.Pipeline;
import com.alibaba.fastjson.JSON;

// JsonPipeline - JSON 文件输出
public class JsonPipeline implements Pipeline {
    private String path;

    public JsonPipeline(String path) {
        this.path = path;
    }

    @Override
    public void process(ResultItems resultItems, Task task) {
        if (resultItems.get("title") != null) {
            try (FileWriter writer = new FileWriter(path, true)) {
                writer.write(JSON.toJSONString(resultItems.getAll()) + "\n");
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }
}

// MongoPipeline - MongoDB 存储
public class MongoPipeline implements Pipeline {
    private MongoCollection<Document> collection;

    public MongoPipeline(String uri, String db, String collection) {
        MongoClient client = MongoClients.create(uri);
        this.collection = client.getDatabase(db).getCollection(collection);
    }

    @Override
    public void process(ResultItems resultItems, Task task) {
        Document doc = new Document(resultItems.getAll());
        collection.insertOne(doc);
    }
}

Scheduler

Scheduler 管理待爬取 URL 的队列和去重。

java
import us.codecraft.webmagic.scheduler.*;

// 内存队列（默认）
Scheduler memoryScheduler = new QueueScheduler();

// 文件持久化队列（可暂停恢复）
Scheduler fileScheduler = new FileCacheQueueScheduler("/data/crawl_queue");

// 优先级队列（先处理高优先级 URL）
Scheduler priorityScheduler = new PriorityScheduler();

// Redis 分布式队列
// Scheduler redisScheduler = new RedisScheduler("localhost:6379");

Spider 启动

java
import us.codecraft.webmagic.Spider;

// 创建并启动 Spider
Spider.create(new ProductPageProcessor())
        .addUrl("https://example.com/products")
        .addPipeline(new JsonPipeline("/data/products.json"))
        .addPipeline(new MongoPipeline("mongodb://localhost:27017", "crawl", "products"))
        .setScheduler(new QueueScheduler())
        .thread(5)                                      // 5 个线程
        .setExitWhenComplete(true)
        .start();

// 断点续爬
Spider spider = Spider.create(new ProductPageProcessor())
        .addUrl("https://example.com/products")
        .setScheduler(new FileCacheQueueScheduler("/data/crawl_queue"))
        .thread(3);

// 手动启停
spider.run();                                           // 阻塞运行
spider.start();                                         // 异步启动
spider.stop();                                          // 停止

HttpClient ​

Apache HttpClient 是 Java 最成熟的 HTTP 客户端库，提供了精细的请求/连接控制。

PoolingHttpClientConnectionManager

连接池管理复用 HTTP 连接，提高爬取效率。

java
import org.apache.http.impl.conn.PoolingHttpClientConnectionManager;
import org.apache.http.impl.client.HttpClients;
import org.apache.http.impl.client.CloseableHttpClient;

// 连接池配置
PoolingHttpClientConnectionManager connectionManager =
        new PoolingHttpClientConnectionManager();

connectionManager.setMaxTotal(200);                  // 最大总连接数
connectionManager.setDefaultMaxPerRoute(50);         // 每个路由最大连接数
connectionManager.setValidateAfterInactivity(2000);  // 空闲 2s 后验证连接

// 创建 HttpClient
CloseableHttpClient httpClient = HttpClients.custom()
        .setConnectionManager(connectionManager)
        .evictIdleConnections(30, TimeUnit.SECONDS)  // 30s 空闲连接回收
        .evictExpiredConnections()                    // 回收过期连接
        .build();

CredentialsProvider

java
import org.apache.http.auth.AuthScope;
import org.apache.http.auth.UsernamePasswordCredentials;
import org.apache.http.client.CredentialsProvider;
import org.apache.http.impl.client.BasicCredentialsProvider;

// 基本认证
CredentialsProvider credsProvider = new BasicCredentialsProvider();
credsProvider.setCredentials(
    AuthScope.ANY,
    new UsernamePasswordCredentials("username", "password")
);

// 代理认证
credsProvider.setCredentials(
    new AuthScope("proxy.example.com", 8080),
    new UsernamePasswordCredentials("proxy_user", "proxy_pass")
);

CloseableHttpClient httpClient = HttpClients.custom()
        .setDefaultCredentialsProvider(credsProvider)
        .build();

RequestConfig

RequestConfig 控制单个请求的超时、重定向、代理等行为。

java
import org.apache.http.client.config.RequestConfig;

// 请求配置
RequestConfig config = RequestConfig.custom()
        .setConnectTimeout(5000)                    // 连接超时（ms）
        .setConnectionRequestTimeout(3000)          // 从连接池获取连接的超时
        .setSocketTimeout(10000)                    // 读取超时（ms）
        .setRedirectsEnabled(true)                  // 自动跟随重定向
        .setMaxRedirects(5)                         // 最大重定向次数
        .setRelativeRedirectsAllowed(true)          // 允许相对路径重定向
        .setProxy(new HttpHost("127.0.0.1", 8080)) // 代理
        .setCircularRedirectsAllowed(false)         // 禁止循环重定向
        .setCookieSpec(CookieSpecs.STANDARD)        // Cookie 规范
        .build();

CloseableHttpClient httpClient = HttpClients.custom()
        .setDefaultRequestConfig(config)
        .build();

CookieStore

java
import org.apache.http.client.CookieStore;
import org.apache.http.impl.client.BasicCookieStore;
import org.apache.http.cookie.Cookie;
import org.apache.http.impl.cookie.BasicClientCookie;

// Cookie 存储
CookieStore cookieStore = new BasicCookieStore();

// 手动设置 Cookie
BasicClientCookie cookie = new BasicClientCookie("session_id", "abc123");
cookie.setDomain(".example.com");
cookie.setPath("/");
cookie.setSecure(false);
cookie.setExpiryDate(new Date(System.currentTimeMillis() + 86400000)); // 1 天后过期
cookieStore.addCookie(cookie);

// 从响应中自动提取 Cookie
CloseableHttpClient httpClient = HttpClients.custom()
        .setDefaultCookieStore(cookieStore)
        .build();

// 获取所有 Cookie
List<Cookie> cookies = cookieStore.getCookies();
for (Cookie c : cookies) {
    System.out.printf("%s = %s (domain: %s)%n", c.getName(), c.getValue(), c.getDomain());
}

// 清除 Cookie
cookieStore.clear();

CloseableHttpClient 完整示例

java
import org.apache.http.client.methods.CloseableHttpResponse;
import org.apache.http.client.methods.HttpGet;
import org.apache.http.client.methods.HttpPost;
import org.apache.http.entity.StringEntity;
import org.apache.http.util.EntityUtils;
import org.apache.http.impl.client.CloseableHttpClient;
import org.apache.http.impl.client.HttpClients;

public class HttpClientExample {
    private final CloseableHttpClient httpClient;
    private final CookieStore cookieStore;

    public HttpClientExample() {
        PoolingHttpClientConnectionManager cm = new PoolingHttpClientConnectionManager();
        cm.setMaxTotal(100);
        cm.setDefaultMaxPerRoute(20);

        this.cookieStore = new BasicCookieStore();
        this.httpClient = HttpClients.custom()
                .setConnectionManager(cm)
                .setDefaultCookieStore(cookieStore)
                .setDefaultRequestConfig(RequestConfig.custom()
                    .setConnectTimeout(5000)
                    .setSocketTimeout(10000)
                    .build())
                .build();
    }

    public String doGet(String url) throws IOException {
        HttpGet request = new HttpGet(url);
        request.setHeader("User-Agent", "Mozilla/5.0 ...");
        request.setHeader("Accept-Language", "zh-CN,zh;q=0.9");

        try (CloseableHttpResponse response = httpClient.execute(request)) {
            // 自动处理 Cookie（通过 CookieStore）
            return EntityUtils.toString(response.getEntity(), "UTF-8");
        }
    }

    public String doPost(String url, String jsonBody) throws IOException {
        HttpPost request = new HttpPost(url);
        request.setHeader("Content-Type", "application/json");
        request.setEntity(new StringEntity(jsonBody, "UTF-8"));

        try (CloseableHttpResponse response = httpClient.execute(request)) {
            return EntityUtils.toString(response.getEntity(), "UTF-8");
        }
    }

    public void close() throws IOException {
        httpClient.close();
    }
}

数据存储与调度 ​

数据存储 ​

存储方式对比表

特性JSONCSVMySQLMongoDB
数据类型结构化/半结构化表格数据结构化文档型
查询能力无无SQL 强大灵活查询
扩展性文件系统文件系统垂直扩展为主水平扩展
写入速度快快慢（事务开销）中
适合数据量小-中小大大-超大
Schema无有严格灵活
关系支持无无强弱（嵌入/引用）
使用场景结构化导出、API 响应数据分析、Excel 导入事务型系统爬虫数据、日志

JSON 存储

python
import json
from datetime import datetime

class JsonStorage:
    def __init__(self, filepath):
        self.filepath = filepath

    def append(self, record):
        record['crawl_time'] = datetime.now().isoformat()
        with open(self.filepath, 'a', encoding='utf-8') as f:
            f.write(json.dumps(record, ensure_ascii=False) + '\n')

    def read_all(self):
        records = []
        with open(self.filepath, 'r', encoding='utf-8') as f:
            for line in f:
                if line.strip():
                    records.append(json.loads(line))
        return records
java
import com.fasterxml.jackson.databind.ObjectMapper;
import java.io.*;
import java.nio.file.*;

public class JsonStorage {
    private final String filepath;
    private final ObjectMapper mapper = new ObjectMapper();

    public JsonStorage(String filepath) {
        this.filepath = filepath;
    }

    public void append(Object record) throws IOException {
        String json = mapper.writeValueAsString(record) + System.lineSeparator();
        Files.write(Paths.get(filepath), json.getBytes(),
                StandardOpenOption.CREATE, StandardOpenOption.APPEND);
    }
}

CSV 存储

python
import csv

class CsvStorage:
    def __init__(self, filepath, fieldnames):
        self.filepath = filepath
        self.fieldnames = fieldnames

    def write_header(self):
        with open(self.filepath, 'w', newline='', encoding='utf-8-sig') as f:
            writer = csv.DictWriter(f, fieldnames=self.fieldnames)
            writer.writeheader()

    def append(self, row):
        with open(self.filepath, 'a', newline='', encoding='utf-8-sig') as f:
            writer = csv.DictWriter(f, fieldnames=self.fieldnames)
            writer.writerow(row)

MySQL 存储 (SQLAlchemy)

python
from sqlalchemy import create_engine, Column, String, Float, DateTime, Integer, Text
from sqlalchemy.orm import declarative_base, sessionmaker
from datetime import datetime

Base = declarative_base()

class Product(Base):
    __tablename__ = 'products'

    id = Column(Integer, primary_key=True, autoincrement=True)
    name = Column(String(255), nullable=False, index=True)
    price = Column(Float, default=0.0)
    url = Column(String(512), unique=True)
    description = Column(Text)
    crawl_time = Column(DateTime, default=datetime.now)

# 创建数据库连接
engine = create_engine('mysql+pymysql://user:pass@localhost:3306/crawldb')
Base.metadata.create_all(engine)
Session = sessionmaker(bind=engine)
session = Session()

# 插入数据
product = Product(name='Example', price=19.99, url='https://example.com/p/1')
session.add(product)
session.commit()

# 批量插入
products = [
    Product(name=f'Product {i}', price=i * 10.0, url=f'https://example.com/p/{i}')
    for i in range(100)
]
session.bulk_save_objects(products)
session.commit()

# 查询
results = session.query(Product).filter(Product.price > 50).order_by(Product.price.desc()).limit(20).all()

MongoDB 存储

python
import pymongo
from pymongo import MongoClient, UpdateOne

class MongoStorage:
    def __init__(self, uri='mongodb://localhost:27017', db_name='crawldb'):
        self.client = MongoClient(uri)
        self.db = self.client[db_name]

    def insert_one(self, collection, data):
        data['crawl_time'] = datetime.now()
        return self.db[collection].insert_one(data)

    def insert_many(self, collection, data_list):
        for d in data_list:
            d['crawl_time'] = datetime.now()
        return self.db[collection].insert_many(data_list)

    def upsert(self, collection, query, data):
        """按条件更新或插入"""
        data['update_time'] = datetime.now()
        return self.db[collection].update_one(query, {'$set': data}, upsert=True)

    def bulk_upsert(self, collection, operations):
        """批量去重写入"""
        requests = []
        for query, data in operations:
            data['update_time'] = datetime.now()
            requests.append(UpdateOne(query, {'$set': data}, upsert=True))
        if requests:
            return self.db[collection].bulk_write(requests)

    def find(self, collection, query=None, limit=100):
        return list(self.db[collection].find(query or {}).limit(limit))

# 使用示例
mongo = MongoStorage()
mongo.upsert('products', {'url': 'https://example.com/p/1'}, {
    'name': 'Example Product',
    'price': 29.99,
    'url': 'https://example.com/p/1',
})

增量爬取 ​

增量爬取的核心是避免重复爬取已处理的数据，只爬取新增或更新的内容。

时间戳策略

记录最后一次爬取时间，仅爬取该时间之后更新的内容。

python
import time
from datetime import datetime, timedelta

class TimestampIncremental:
    def __init__(self, state_file='crawl_state.json'):
        self.state_file = state_file
        self.last_crawl_time = self.load_state()

    def load_state(self):
        try:
            with open(self.state_file, 'r') as f:
                state = json.load(f)
                return datetime.fromisoformat(state['last_crawl_time'])
        except:
            return datetime.now() - timedelta(days=7)

    def save_state(self):
        with open(self.state_file, 'w') as f:
            json.dump({'last_crawl_time': datetime.now().isoformat()}, f)

    def should_crawl(self, item_update_time):
        """判断是否需要爬取"""
        return item_update_time > self.last_crawl_time

指纹去重

通过计算 URL 或内容特征的哈希值进行去重。

python
import hashlib

class FingerprintDeduplicator:
    def __init__(self):
        self.fingerprints = set()

    def url_fingerprint(self, url):
        return hashlib.md5(url.encode('utf-8')).hexdigest()

    def content_fingerprint(self, content):
        return hashlib.sha256(content.encode('utf-8')).hexdigest()

    def is_duplicate(self, url, content=None):
        fp = self.url_fingerprint(url)
        if fp in self.fingerprints:
            return True
        self.fingerprints.add(fp)
        return False

Redis 集合去重

使用 Redis 的 Set 数据结构实现分布式环境下的去重。

python
import redis

class RedisDeduplicator:
    def __init__(self, redis_url='redis://localhost:6379', key='crawler:visited'):
        self.client = redis.from_url(redis_url)
        self.key = key

    def add(self, url):
        fp = hashlib.md5(url.encode()).hexdigest()
        return self.client.sadd(self.key, fp)

    def is_visited(self, url):
        fp = hashlib.md5(url.encode()).hexdigest()
        return self.client.sismember(self.key, fp)

    def is_new(self, url):
        """如果是新 URL 则添加并返回 True，否则返回 False"""
        return bool(self.add(url))

    def count(self):
        return self.client.scard(self.key)

    def clear(self):
        self.client.delete(self.key)

BloomFilter 去重

布隆过滤器使用概率性数据结构，占用极少内存即可处理海量 URL 去重。

python
import hashlib
import math
from bitarray import bitarray

class BloomFilter:
    def __init__(self, capacity=1_000_000, error_rate=0.001):
        """
        capacity: 预期插入元素数量
        error_rate: 误判率
        """
        self.capacity = capacity
        self.error_rate = error_rate
        self.bit_size = int(-capacity * math.log(error_rate) / (math.log(2) ** 2))
        self.hash_count = int(self.bit_size / capacity * math.log(2))
        self.bit_array = bitarray(self.bit_size)
        self.bit_array.setall(0)

    def _hashes(self, item):
        """生成多个哈希值"""
        md5 = hashlib.md5(item.encode()).hexdigest()
        sha1 = hashlib.sha1(item.encode()).hexdigest()
        digests = []
        for i in range(self.hash_count):
            combined = f'{md5}{sha1}{i}'
            h = int(hashlib.sha256(combined.encode()).hexdigest(), 16)
            digests.append(h % self.bit_size)
        return digests

    def add(self, item):
        for h in self._hashes(item):
            self.bit_array[h] = 1

    def contains(self, item):
        for h in self._hashes(item):
            if not self.bit_array[h]:
                return False
        return True

# 使用 Redis 实现的分布式 BloomFilter
class RedisBloomFilter:
    def __init__(self, redis_url='redis://localhost:6379', key='crawler:bloom'):
        self.client = redis.from_url(redis_url)
        self.key = key

    def add(self, url):
        fp = hashlib.md5(url.encode()).hexdigest()
        # 使用 Redis 的 SETBIT
        for i in range(7):  # 7 个哈希位置
            self.client.setbit(self.key, hash(fp + str(i)) % (10**8), 1)

    def is_visited(self, url):
        fp = hashlib.md5(url.encode()).hexdigest()
        for i in range(7):
            if not self.client.getbit(self.key, hash(fp + str(i)) % (10**8)):
                return False
        return True

调度管理 ​

定时爬取 APScheduler

APScheduler 是 Python 最常用的定时任务库，支持多种触发器。

python
from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.triggers.cron import CronTrigger
from apscheduler.triggers.interval import IntervalTrigger
from apscheduler.executors.pool import ThreadPoolExecutor

def crawl_task():
    """爬虫任务函数"""
    print(f"[{datetime.now()}] 开始爬取...")
    # 执行爬虫逻辑
    spider = ProductSpider()
    spider.run()
    print(f"[{datetime.now()}] 爬取完成")

# 创建调度器
scheduler = BackgroundScheduler(
    executors={'default': ThreadPoolExecutor(10)},
    job_defaults={
        'coalesce': True,       # 积压任务合并执行
        'max_instances': 1,     # 同一任务最多一个实例
        'misfire_grace_time': 300,  # 错过执行时间的宽限秒数
    }
)

# Cron 表达式（每天上午 9 点和下午 6 点执行）
scheduler.add_job(
    crawl_task,
    CronTrigger(hour='9,18', minute=0),
    id='crawl_daily',
    replace_existing=True,
)

# 间隔触发（每 30 分钟执行一次）
scheduler.add_job(
    crawl_task,
    IntervalTrigger(minutes=30),
    id='crawl_interval',
)

# 启动调度器
scheduler.start()

Cron 表达式

字段允许值特殊字符
秒0-59, - * /
分0-59, - * /
时0-23, - * /
日1-31, - * / ? L W
月1-12 或 JAN-DEC, - * /
周0-6 或 SUN-SAT, - * / ? L #

常用 Cron 表达式示例：

表达式说明
0 0 * * *每小时执行
0 0/2 * * *每 2 小时执行
0 0 9 * * *每天上午 9 点
0 30 9 * * 1-5工作日 9:30
0 0 0 1 * *每月 1 号凌晨
0 0 3 * * 0每周日凌晨 3 点

Celery Beat 定时任务

Celery Beat 是 Celery 的周期性任务调度器，适合大规模分布式爬虫场景。

python
# celery_app.py
from celery import Celery
from celery.schedules import crontab

app = Celery('crawler',
    broker='redis://localhost:6379/0',
    backend='redis://localhost:6379/1',
)

app.conf.update(
    task_serializer='json',
    accept_content=['json'],
    result_serializer='json',
    timezone='Asia/Shanghai',
    enable_utc=False,
    task_acks_late=True,          # 任务完成后才确认
    worker_prefetch_multiplier=1, # 每次只预取一个任务
    task_reject_on_worker_lost=True,
)

# 定义爬虫任务
@app.task(bind=True, max_retries=3, default_retry_delay=60)
def crawl_category(self, category_id):
    try:
        # 爬虫逻辑
        pass
    except Exception as e:
        self.retry(exc=e)

@app.task(bind=True, max_retries=5, default_retry_delay=120)
def crawl_product(self, product_url):
    try:
        # 爬取商品详情
        pass
    except requests.Timeout:
        self.retry(exc=e, countdown=30)

# 定时任务配置
app.conf.beat_schedule = {
    'crawl-hot-categories': {
        'task': 'crawler.tasks.crawl_category',
        'schedule': crontab(minute=0, hour='*/2'),
        'args': (1,),
    },
    'crawl-all-products': {
        'task': 'crawler.tasks.crawl_product',
        'schedule': crontab(minute=0, hour=3),
        'args': ('https://example.com/sitemap.xml',),
    },
}

# 启动：celery -A celery_app worker -l info --beat

任务优先级

python
# Celery 任务优先级
app.conf.task_queue_max_priority = 10

@app.task(priority=10)
def high_priority_task():
    """高优先级任务"""
    pass

@app.task(priority=1)
def low_priority_task():
    """低优先级任务"""
    pass

# Scrapy 请求优先级
from scrapy.http import Request

# 高优先级（数值越大优先级越高）
yield Request(url, callback=self.parse_detail, priority=10)
# 低优先级
yield Request(url, callback=self.parse, priority=0)
# 默认优先级
yield Request(url, callback=self.parse)

失败重试与爬取队列

python
import queue
import time
from dataclasses import dataclass
from typing import Callable

@dataclass
class CrawlTask:
    url: str
    callback: Callable
    priority: int = 0
    retry_count: int = 0
    max_retries: int = 3
    error: Exception = None

class CrawlQueue:
    """带优先级和重试机制的爬取队列"""
    def __init__(self):
        self.queue = queue.PriorityQueue()
        self.failed_tasks = []
        self.running = False

    def add_task(self, task: CrawlTask):
        """添加任务（优先级取反实现大值优先）"""
        self.queue.put((-task.priority, task))

    def add_failed(self, task: CrawlTask):
        """处理失败任务"""
        task.retry_count += 1
        if task.retry_count < task.max_retries:
            # 指数退避
            delay = 2 ** task.retry_count
            print(f"任务 {task.url} 失败，{delay}s 后重试 (第 {task.retry_count} 次)")
            task.priority = -task.retry_count  # 降低优先级
            self.add_task(task)
        else:
            print(f"任务 {task.url} 已达到最大重试次数，放弃")
            self.failed_tasks.append(task)

    def run(self, workers=5):
        """运行爬取队列"""
        self.running = True
        while self.running:
            try:
                _, task = self.queue.get(timeout=5)
                try:
                    task.callback(task.url)
                except Exception as e:
                    task.error = e
                    self.add_failed(task)
                self.queue.task_done()
            except queue.Empty:
                if self._should_stop():
                    self.running = False

    def _should_stop(self):
        return self.queue.empty() and not self.failed_tasks

    def stats(self):
        return {
            'queue_size': self.queue.qsize(),
            'failed_count': len(self.failed_tasks),
        }