当前位置：首页 > news >正文

从零手撸OpenClaw：理解网络爬虫核心原理与轻量级框架实现

news 2026/8/13 10:48:27

1. 从“手撸”到“OpenClaw”：一个程序员的浪漫

最近在技术社区里，看到不少朋友在讨论“手撸”这个词。它早已超越了字面意思，成了一种极客精神的代名词：不依赖现成的重型框架，从最核心的原理出发，自己动手实现一个功能或工具。这过程就像亲手组装一台模型，或者，更接地气一点，像在夜市大排档里，亲手剥开一只小龙虾——过程充满挑战，但最终收获的成就感和对“食材”的理解，是直接享用成品无法比拟的。

今天，我们就来“手撸”一个名为OpenClaw的小工具。这个名字听起来有点酷，又带点趣味。“Claw”是爪子，在编程世界里，它常常指代那些能“抓取”东西的工具，比如网络爬虫。所以，OpenClaw 本质上是一个轻量级、可扩展的网络爬虫框架。我们不用 Scrapy 这样的“工业级起重机”，也不用 Puppeteer 这样的“自动化机械臂”，而是用最基础的“螺丝刀”和“扳手”（即 Python 标准库和少量第三方库），从零开始搭建一个能抓取网页、解析数据、并保存结果的核心骨架。

为什么要在 Scrapy 等成熟框架大行其道的今天，还要做这件事？原因有三。第一，理解本质：框架封装了复杂性，也隐藏了细节。自己实现一遍，你会彻底明白 HTTP 请求、HTML 解析、异步并发、去重策略这些爬虫核心组件是如何协同工作的。第二，极致定制：当你的需求非常特殊，或者对性能、资源有极端要求时，一个量身定做、没有冗余的“手撸”工具往往比改造一个庞大框架更高效。第三，学习与乐趣：这是最好的编程练习，涉及网络、数据结构、并发编程、设计模式等多个领域，完成后的成就感十足。

接下来，我们将用大约10分钟的核心讲解时间（实际动手编码和思考的时间当然会更长），一步步构建 OpenClaw 的雏形。目标是让它具备最核心的抓取能力，结构清晰，方便后续按需添加代理、分布式、反反爬虫等高级特性。无论你是想深入理解爬虫原理的新手，还是需要为一个轻量级特定任务寻找解决方案的开发者，这篇内容都能给你带来直接的参考价值。

2. OpenClaw 的核心架构设计：四两拨千斤

在动手写代码之前，我们必须先想清楚 OpenClaw 应该长什么样。一个好的设计能让后续的编码事半功倍，也决定了这个工具的扩展上限。我们不追求大而全，而是要精准命中爬虫工作流中最关键的几个环节。

一个最基本的爬虫工作流可以抽象为以下几个步骤：

调度：决定下一个要抓取的网址（URL）是什么。
下载：向目标网址发起 HTTP 请求，获取网页的原始内容（HTML、JSON 等）。
解析：从下载的原始内容中，提取出我们感兴趣的结构化数据（如文章标题、价格、链接）。
处理与存储：对提取的数据进行清洗、验证，然后保存到文件或数据库。
链接提取：从已解析的页面中，发现新的、需要抓取的网址，并送回给调度器。

基于这个流程，我们可以为 OpenClaw 设计一个经典的生产者-消费者模型，核心包含四个模块：

2.1 调度器：任务队列的管理者

调度器是爬虫的大脑，负责管理待抓取的 URL 队列。我们需要考虑几个关键问题：

队列数据结构：Python 的collections.deque（双端队列）是一个很好的起点，它支持高效的头部弹出和尾部追加。对于更复杂的优先级调度，可以使用heapq模块实现优先队列。
去重：绝对不能重复抓取同一个 URL。我们将在调度器内部维护一个“已访问集合”。一个简单有效的去重方法是使用set()存储 URL，但对于海量 URL，内存可能吃紧。因此，我们的设计要留有接口，未来可以轻松替换为基于布隆过滤器或数据库的方案。
线程安全：如果我们的下载器是多线程的，那么对任务队列的存取操作必须是原子的。我们可以直接使用queue.Queue，它天生就是线程安全的。

在 OpenClaw 的初版中，我们将采用queue.Queue作为任务队列，并用一个set做内存去重，实现简单直观。

2.2 下载器：与网络对话的工人

下载器的唯一职责就是根据 URL 下载内容。这里面的门道不少：

请求库选择：requests库是同步请求的绝佳选择，简单易用。但对于高性能爬虫，异步是必由之路。aiohttp是异步 HTTP 客户端的标杆。为了平衡难度和实用性，我们的 OpenClaw 初版将使用requests，但会以面向接口的方式编写下载器，以便后续无缝切换为aiohttp。
请求头模拟：这是绕过基础反爬虫的第一关。一个没有User-Agent的请求就像没穿衣服上街，异常显眼。我们需要让下载器能够接收并应用自定义的请求头。
异常处理与重试：网络世界充满不确定性：连接超时、服务器错误、页面不存在。一个健壮的下载器必须包含异常捕获和重试机制。我们可以设计一个装饰器或直接在方法内部实现带指数退避的重试逻辑。
延迟控制：尊重目标网站，避免请求过快被封 IP。下载器应该支持在每个请求之间插入随机延迟。

我们将实现一个Downloader类，它接收一个 URL 和请求头，返回响应文本和状态码，并内置简单的重试和延迟功能。

2.3 解析器：从混沌中提取秩序

下载到的是混杂着标签、样式和脚本的 HTML“矿石”，解析器的任务就是从中提炼出数据的“金属”。

解析库选择：BeautifulSoup4配合lxml解析引擎是 Python 界的事实标准，写法接近自然语言，非常友好。对于极致的性能，可以直接使用lxml的 XPath 或parsel库（Scrapy 在用）。我们选择BeautifulSoup4作为起点，因为它的学习曲线平缓，足以应对 90% 的场景。
解析规则抽象：我们不能把解析逻辑硬编码在爬虫主体里。好的设计是将解析规则独立出来，可能是一个函数、一个配置字典，甚至一个类。这样，同一个爬虫框架就能通过更换解析规则来抓取不同的网站。
数据封装：解析出的数据应该被封装成结构化的对象，比如 Python 字典或dataclass。这有利于后续的数据处理和存储。

我们将定义一个Parser基类，要求子类实现一个parse方法。这个方法接收 HTML 文本和当前 URL，返回两项内容：一是提取到的结构化数据项列表，二是从当前页面中发现的新 URL 列表。

2.4 数据管道：数据的终点站

数据管道负责处理解析器产出的数据。它可能做很多事情：

数据清洗：去除空白字符、转换格式、处理缺失值。
数据验证：检查必填字段是否存在，数据格式是否正确。
数据存储：将数据保存到 CSV 文件、JSON 文件、MySQL、MongoDB 等。和解析器一样，数据管道也应该被设计成可插拔的组件。我们可以定义一个Pipeline基类，子类实现process_item方法。

有了以上设计，OpenClaw 的主循环逻辑就清晰了：调度器出队一个 URL -> 下载器下载 -> 解析器解析（得到数据和新的URL）-> 新URL送回调度器入队 -> 数据送入管道处理。这个循环直到任务队列为空或达到其他停止条件为止。

3. 手把手实现 OpenClaw 核心模块

理论说得够多了，现在打开你的代码编辑器，我们开始真正“手撸”。我们将按照架构，逐个模块实现。请确保你已安装必要的库：pip install requests beautifulsoup4。

3.1 构建调度器：简单而稳固的队列核心

首先，我们实现调度器Scheduler。它需要完成添加任务、获取任务、判断任务是否已存在等基本功能。

import queue from typing import Set from urllib.parse import urljoin class Scheduler: def __init__(self): # 使用线程安全的队列存储待抓取URL self.task_queue = queue.Queue() # 使用集合存储已看到过的URL，用于去重 self.seen_urls: Set[str] = set() def add_task(self, url: str, base_url: str = None): """添加一个新任务到队列。""" # 可选：将相对URL转换为绝对URL if base_url and not url.startswith(('http://', 'https://')): url = urljoin(base_url, url) # 去重检查：如果URL没被见过，则加入队列和集合 if url not in self.seen_urls: self.seen_urls.add(url) self.task_queue.put(url) print(f"[调度器] 新任务入队: {url}") def get_task(self): """从队列中获取一个任务。如果队列为空，返回None。""" try: # block=False 表示非阻塞获取，队列空时立刻抛出Empty异常 return self.task_queue.get(block=False) except queue.Empty: return None def has_pending_tasks(self): """判断是否还有待处理的任务。""" return not self.task_queue.empty()

关键点解析：

urljoin的作用：网页中的链接常常是相对路径（如/article/123）。urljoin能根据当前页面的base_url，将其补全为绝对 URL（如https://example.com/article/123），这是爬虫能持续工作的基础。
去重时机：我们在add_task时就去重，而不是在get_task时。这保证了队列中本身就没有重复项，更清晰。seen_urls集合保存在内存中，适用于中小规模抓取。
队列选择：queue.Queue的get(block=False)让我们可以在主循环中非阻塞地检查是否有新任务，方便控制爬虫的启停。

3.2 打造下载器：稳健的网络请求客户端

接下来是下载器Downloader。它要处理网络请求的所有细节。

import time import random from typing import Optional import requests from requests.exceptions import RequestException class Downloader: def __init__(self, delay: float = 1.0, max_retries: int = 3): """ 初始化下载器。 :param delay: 请求之间的基础延迟（秒），用于礼貌爬取。 :param max_retries: 请求失败时的最大重试次数。 """ self.delay = delay self.max_retries = max_retries # 创建一个共享的Session，可以复用TCP连接，提升效率 self.session = requests.Session() # 设置一个默认的User-Agent，模拟常见浏览器 self.default_headers = { 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36' } def download(self, url: str, headers: Optional[dict] = None) -> Optional[str]: """ 下载给定URL的内容。 :return: 成功则返回网页文本，失败则返回None。 """ # 合并默认头和自定义头 final_headers = {**self.default_headers, **(headers or {})} # 重试逻辑 for attempt in range(self.max_retries): try: # 请求前等待，避免过快 time.sleep(self.delay + random.uniform(0, 0.5)) # 增加一点随机性 print(f"[下载器] 尝试抓取 ({attempt+1}/{self.max_retries}): {url}") response = self.session.get(url, headers=final_headers, timeout=10) response.raise_for_status() # 如果状态码不是200，抛出HTTPError异常 # 检查编码，避免乱码 response.encoding = response.apparent_encoding return response.text except RequestException as e: print(f"[下载器] 请求失败: {e}") if attempt == self.max_retries - 1: # 最后一次尝试也失败了 return None # 指数退避：失败后等待更长时间再重试 wait_time = 2 ** attempt print(f"[下载器] 等待 {wait_time} 秒后重试...") time.sleep(wait_time) return None

关键点解析与避坑指南：

使用 Session：requests.Session()可以保持 cookies，并在多个请求间复用底层 TCP 连接，显著减少网络开销，这是提升性能的一个小技巧。
response.raise_for_status()：这个方法非常有用。requests.get()即使遇到 404、500 等错误状态码，也不会抛出异常（除非你设置raise_for_status=True）。手动调用此方法可以确保只处理成功的响应，将错误处理统一到except块中。
编码处理：网页编码千奇百怪。response.apparent_encoding是requests库基于内容分析出的可能编码，比response.encoding（来自HTTP头）通常更可靠，能有效避免中文乱码问题。
延迟与随机性：固定的延迟（如time.sleep(1)）模式容易被识别。加入随机扰动（random.uniform(0, 0.5)）能使爬虫行为更接近人类，是基础的“反反爬虫”策略。
指数退避重试：网络请求失败是常态。指数退避策略（等待 1秒、2秒、4秒...）既能给服务器喘息之机，也能提高在临时网络波动下最终成功的概率。

3.3 实现解析器：定义数据提取的规则

解析器我们设计成抽象基类，具体规则由用户实现。这里我们以一个抓取虚构新闻网站为例。

from abc import ABC, abstractmethod from typing import List, Tuple from bs4 import BeautifulSoup class Parser(ABC): """解析器抽象基类。所有具体的解析规则都应继承此类。""" @abstractmethod def parse(self, html: str, current_url: str) -> Tuple[List[dict], List[str]]: """ 解析HTML，提取数据和新的链接。 :param html: 下载的HTML文本。 :param current_url: 当前页面的URL，用于将相对链接转为绝对链接。 :return: (提取到的数据项列表, 新发现的URL列表) """ pass class NewsParser(Parser): """一个示例解析器：用于解析一个简单的新闻列表页和详情页。""" def parse(self, html: str, current_url: str) -> Tuple[List[dict], List[str]]: data_items = [] new_urls = [] soup = BeautifulSoup(html, 'lxml') # 情况1：当前页面是列表页（包含多个新闻链接） # 假设列表页中，新闻链接在 <a class="news-link"> 标签里 for link_tag in soup.find_all('a', class_='news-link'): href = link_tag.get('href') if href: new_urls.append(href) # 这里先存相对链接，调度器会处理 # 也可以在列表页预提取一些信息，比如标题 title = link_tag.get_text(strip=True) if title: data_items.append({'title': title, 'url': href, 'source_page': 'list'}) # 情况2：当前页面是详情页（包含新闻正文） # 假设详情页中，标题在 <h1 id="article-title">，正文在 <div class="article-content"> title_tag = soup.find('h1', id='article-title') content_div = soup.find('div', class_='article-content') if title_tag and content_div: # 如果找到这两个元素，则认为这是详情页 title = title_tag.get_text(strip=True) content = content_div.get_text(strip=True, separator='\n') # 用换行符保留段落感 data_items.append({ 'title': title, 'content': content, 'url': current_url, 'source_page': 'detail' }) # 详情页可能也有相关文章链接，可以继续提取 for related_link in soup.select('.related-news a'): href = related_link.get('href') if href: new_urls.append(href) # 情况3：提取页面中所有其他符合条件的链接（广度爬取） # 例如，抓取分页链接。假设分页链接在 <a class="page-link"> 里 for page_link in soup.find_all('a', class_='page-link'): href = page_link.get('href') if href and href not in new_urls: # 简单去重 new_urls.append(href) return data_items, new_urls

关键点解析：

抽象基类：使用ABC和@abstractmethod强制要求子类实现parse方法。这定义了框架和插件之间的契约，使得 OpenClaw 的核心循环可以调用parser.parse(...)而无需关心具体实现。
灵活的解析逻辑：NewsParser展示了如何在一个解析器里处理多种页面类型（列表页、详情页）。通过检查 HTML 中是否存在特定元素来判断页面类型，这是一种很实用的模式。
BeautifulSoup方法选择：
- find_all(): 查找所有符合条件的标签。
- find(): 查找第一个符合条件的标签。
- select(): 使用 CSS 选择器查找，功能强大且写法简洁。
- get_text(strip=True, separator='\n'): 获取标签内所有文本，strip=True去除首尾空白，separator可以指定标签间文本的分隔符，对于正文提取非常有用。
链接去重：在解析器内部对提取的新 URL 做一层简单的去重（if href not in new_urls），可以减少调度器的无效工作。但调度器的全局去重仍然是最终保障。

3.4 组装数据管道：数据的清洁工与仓库管理员

数据管道负责处理解析后的数据。我们先实现一个最简单的控制台打印管道和一个 CSV 文件存储管道。

from abc import ABC, abstractmethod import csv import json from typing import List class Pipeline(ABC): """数据管道抽象基类。""" @abstractmethod def process_item(self, item: dict): """处理一个数据项。""" pass def close(self): """管道关闭时执行的操作（如关闭文件句柄、数据库连接）。""" pass class ConsolePipeline(Pipeline): """将数据打印到控制台的管道。用于调试。""" def process_item(self, item: dict): print("[控制台管道] 抓到数据:", json.dumps(item, ensure_ascii=False, indent=2)) class CsvPipeline(Pipeline): """将数据存储到CSV文件的管道。""" def __init__(self, filename: str = 'output.csv'): self.filename = filename self.file = None self.writer = None self.fieldnames = None # CSV文件的表头 def open_spider(self): """在爬虫开始时调用（这里由主循环手动触发）。""" self.file = open(self.filename, 'w', newline='', encoding='utf-8-sig') # utf-8-sig 支持Excel直接打开 # 先不创建writer，等收到第一个item时根据keys确定表头 def process_item(self, item: dict): if self.writer is None: # 第一个item到达，根据其keys创建CSV表头和写入器 self.fieldnames = item.keys() self.writer = csv.DictWriter(self.file, fieldnames=self.fieldnames) self.writer.writeheader() # 写入一行数据 self.writer.writerow(item) print(f"[CSV管道] 数据已写入: {item}") def close(self): if self.file: self.file.close() print(f"[CSV管道] 文件 {self.filename} 已关闭。")

关键点解析：

延迟创建表头：CsvPipeline采用了一种巧妙的设计——在process_item收到第一个数据项时，才根据这个数据项的键来确定 CSV 文件的表头。这使得管道无需预先知道数据的具体结构，非常灵活。无论解析器提取出什么字段，都能自动适应。
utf-8-sig编码：使用utf-8-sig编码写入 CSV 文件，会在文件开头加入一个 BOM（字节顺序标记）。这对于 Windows 系统上的 Excel 等软件非常友好，能确保中文字符正确显示，而不会出现乱码。
管道组合：我们可以很容易地创建多个管道实例（比如一个打印到控制台，一个保存到 CSV，一个发送到数据库），并在主循环中依次调用它们的process_item方法。这就是“管道”模式的威力，数据处理流程清晰且可扩展。

4. 编写主引擎与实战测试：让 OpenClaw 动起来

现在，我们将上面散落的模块组装起来，形成一个完整的爬虫引擎OpenClawEngine，并写一个简单的main函数来测试它。

4.1 构建核心引擎

class OpenClawEngine: """OpenClaw 的核心引擎，负责协调各模块工作。""" def __init__(self, downloader, parser, pipelines=None): self.scheduler = Scheduler() self.downloader = downloader self.parser = parser # pipelines 是一个管道实例的列表 self.pipelines = pipelines or [] def add_start_url(self, url: str): """添加初始URL。""" self.scheduler.add_task(url) def run(self): """启动爬虫主循环。""" print("[引擎] OpenClaw 启动！") # 启动管道（如果有需要初始化的） for pipeline in self.pipelines: if hasattr(pipeline, 'open_spider'): pipeline.open_spider() try: while self.scheduler.has_pending_tasks(): # 1. 调度：获取下一个任务 current_url = self.scheduler.get_task() if not current_url: break print(f"\n[引擎] 处理: {current_url}") # 2. 下载 html = self.downloader.download(current_url) if html is None: print(f"[引擎] 下载失败，跳过: {current_url}") continue # 3. 解析 data_items, new_urls = self.parser.parse(html, current_url) print(f"[引擎] 解析出 {len(data_items)} 条数据， {len(new_urls)} 个新链接") # 4. 新链接送回调度器 for url in new_urls: self.scheduler.add_task(url, base_url=current_url) # 5. 数据送入管道处理 for item in data_items: for pipeline in self.pipelines: pipeline.process_item(item) except KeyboardInterrupt: print("\n[引擎] 用户中断，停止爬取。") finally: # 关闭管道（释放资源） for pipeline in self.pipelines: pipeline.close() print("[引擎] 爬取结束。")

引擎逻辑详解：

清晰的流程：run方法完美体现了我们之前设计的核心循环：调度 -> 下载 -> 解析 -> （链接反馈）-> 数据处理。代码就是设计文档。
资源管理：在try...finally块中，我们确保了即使爬虫被用户强制中断（KeyboardInterrupt），管道也能被正确关闭，文件句柄等资源得到释放。这是编写健壮程序的好习惯。
可扩展性：引擎通过构造函数接收downloader,parser,pipelines。这意味着我们可以轻松地替换其中任何一个组件。例如，明天你想用aiohttp，只需写一个AsyncDownloader类替换掉现在的Downloader实例即可，引擎代码无需改动。

4.2 编写测试用例与模拟服务器

为了在不触及真实网络的情况下测试我们的 OpenClaw，我们需要一个本地模拟服务器。Python 内置的http.server模块可以快速实现。

首先，创建一个名为test_server.py的文件，并在同目录下创建一些模拟的 HTML 文件。

模拟的列表页 (index.html):

<!DOCTYPE html> <html> <head><title>测试新闻站</title></head> <body> <h1>新闻列表</h1> <ul> <li><a class="news-link" href="/article/1">第一条测试新闻</a></li> <li><a class="news-link" href="/article/2">第二条测试新闻</a></li> </ul> <div class="pagination"> <a class="page-link" href="/page/2">下一页</a> </div> </body> </html>

模拟的详情页 (article1.html):

<!DOCTYPE html> <html> <head><title>第一条测试新闻</title></head> <body> <h1 id="article-title">第一条测试新闻</h1> <div class="article-content"> <p>这是第一条测试新闻的详细内容。</p> <p>这里包含了我们想抓取的所有正文信息。</p> </div> <div class="related-news"> <h3>相关新闻</h3> <a href="/article/3">相关新闻一</a> </div> </body> </html>

创建test_server.py:

import http.server import socketserver import os PORT = 8888 DIRECTORY = "." # 当前目录 class Handler(http.server.SimpleHTTPRequestHandler): def __init__(self, *args, **kwargs): super().__init__(*args, directory=DIRECTORY, **kwargs) def log_message(self, format, *args): # 重写此方法以减少控制台输出 pass with socketserver.TCPServer(("", PORT), Handler) as httpd: print(f"测试服务器启动于 http://localhost:{PORT}") print(f"服务目录: {os.path.abspath(DIRECTORY)}") try: httpd.serve_forever() except KeyboardInterrupt: print("\n服务器关闭。")

4.3 运行完整的集成测试

现在，在另一个终端或你的主脚本中，启动 OpenClaw 来抓取我们刚创建的模拟网站。

# main.py if __name__ == '__main__': # 1. 初始化组件 my_downloader = Downloader(delay=0.5) # 测试时延迟调低 my_parser = NewsParser() my_pipelines = [ ConsolePipeline(), CsvPipeline('news_data.csv') ] # 2. 创建引擎 engine = OpenClawEngine( downloader=my_downloader, parser=my_parser, pipelines=my_pipelines ) # 3. 添加种子URL（我们的模拟服务器地址） start_url = "http://localhost:8888/index.html" engine.add_start_url(start_url) # 4. 运行！ engine.run()

运行步骤：

在一个终端，先运行python test_server.py启动本地测试服务器。
在另一个终端，运行python main.py启动 OpenClaw。
观察控制台输出，你会看到调度、下载、解析、存储的完整日志。同时，当前目录下会生成一个news_data.csv文件，里面包含了抓取到的数据。

通过这个完整的测试，你不仅验证了 OpenClaw 各个模块能协同工作，也体验了一个爬虫从 URL 种子开始，如何自动发现新链接、抓取数据并保存的全过程。这种“闭环”测试对于构建任何工具都至关重要。

5. 性能优化与扩展方向：从玩具到工具

我们有了一个能跑起来的 OpenClaw，但它目前还是单线程的“玩具”。要把它变成实用的“工具”，我们需要考虑性能和扩展性。这里提供几个最关键的优化和扩展思路。

5.1 引入并发：从同步到异步

同步的requests库在等待服务器响应时会阻塞整个线程。对于 I/O 密集型的网络爬虫，这是巨大的性能瓶颈。解决方案是使用异步编程。

方案一：多线程/多进程

优点：概念相对简单，利用concurrent.futures模块可以快速实现。
缺点：线程/进程切换有开销，且 Python 的 GIL 限制了多线程的 CPU 并行能力（但爬虫主要是 I/O 等待，影响相对小）。需要小心管理共享资源（如调度器队列）的线程安全。
改造点：将Downloader.download任务提交给线程池。调度器Scheduler必须使用线程安全的queue.Queue（我们已经做了）。需要实现一个工作线程函数，从队列取任务，下载，解析，再投递新任务。

方案二：异步 I/O (asyncio + aiohttp)

优点：这是现代 Python 高性能网络编程的首选。单线程即可处理成千上万个并发连接，资源利用率极高。
缺点：需要理解异步编程范式（async/await），代码改造幅度较大。
改造点：
1. 将Downloader重写为AsyncDownloader，使用aiohttp.ClientSession。
2. 将Scheduler的方法改为async，可能使用asyncio.Queue。
3. 主循环run需要改写成异步的，并使用asyncio.gather或asyncio.create_task来并发执行多个下载任务。
4. 解析和管道处理如果是 CPU 密集型，可以考虑放到线程池中执行，避免阻塞事件循环。

一个简单的多线程改造示意：

from concurrent.futures import ThreadPoolExecutor, as_completed class OpenClawEngine: # ... __init__ 等部分不变 ... def run(self, max_workers=3): # 增加最大工作线程数参数 print("[引擎] OpenClaw 启动！") for pipeline in self.pipelines: if hasattr(pipeline, 'open_spider'): pipeline.open_spider() with ThreadPoolExecutor(max_workers=max_workers) as executor: future_to_url = {} try: # 初始提交一批任务 initial_tasks = [] while len(initial_tasks) < max_workers and self.scheduler.has_pending_tasks(): url = self.scheduler.get_task() if url: initial_tasks.append(url) for url in initial_tasks: future = executor.submit(self._process_one_url, url) future_to_url[future] = url # 处理完成的任务，并提交新任务 while future_to_url: done, _ = concurrent.futures.wait(future_to_url.keys(), return_when=concurrent.futures.FIRST_COMPLETED) for future in done: url = future_to_url.pop(future) try: new_urls_from_task = future.result() # 获取任务结果（新URL列表） for new_url in new_urls_from_task: self.scheduler.add_task(new_url, url) except Exception as e: print(f"处理 {url} 时出错: {e}") # 补充新任务到线程池 if self.scheduler.has_pending_tasks(): next_url = self.scheduler.get_task() if next_url: new_future = executor.submit(self._process_one_url, next_url) future_to_url[new_future] = next_url except KeyboardInterrupt: executor.shutdown(wait=False, cancel_futures=True) print("\n[引擎] 用户中断，停止爬取。") finally: for pipeline in self.pipelines: pipeline.close() print("[引擎] 爬取结束。") def _process_one_url(self, url): """处理单个URL的完整流程，适配线程池调用。""" html = self.downloader.download(url) if html is None: return [] data_items, new_urls = self.parser.parse(html, url) for item in data_items: for pipeline in self.pipelines: pipeline.process_item(item) return new_urls # 返回新URL列表供主线程加入调度器

注意：这是一个高度简化的多线程示例，真实场景下需要更精细的任务调度、错误处理和队列管理。但它清晰地展示了将阻塞的下载任务放到线程池中的思路。

5.2 应对反爬虫策略：提升生存能力

真实的网站会有反爬虫措施。我们的 OpenClaw 需要一些“盔甲”。

User-Agent 轮换：准备一个 User-Agent 列表，每次请求随机选择一个。

class SmartDownloader(Downloader): def __init__(self, delay=1.0, max_retries=3): super().__init__(delay, max_retries) self.user_agents = [ 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 ...', 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/605.1.15 ...', # ... 更多UA ] def download(self, url, headers=None): final_headers = {**self.default_headers, **(headers or {})} final_headers['User-Agent'] = random.choice(self.user_agents) # 随机选择 # ... 其余逻辑不变

请求头完善：模拟更真实的浏览器，添加Accept、Accept-Language、Referer（可设置为当前页面的来源页）等头部信息。
Cookie 管理：对于需要登录的网站，使用requests.Session()可以自动管理 Cookie。对于更复杂的场景，可能需要从浏览器导出 Cookie 文件并加载。

IP 代理池：当单个 IP 请求过于频繁被封锁时，需要使用代理 IP。可以集成第三方代理服务，或者在下载器中加入代理切换逻辑。

class ProxiedDownloader(Downloader): def __init__(self, proxy_list=None, **kwargs): super().__init__(**kwargs) self.proxy_list = proxy_list or [] self.proxy_index = 0 def get_proxy(self): if not self.proxy_list: return None proxy = self.proxy_list[self.proxy_index] self.proxy_index = (self.proxy_index + 1) % len(self.proxy_list) return {'http': proxy, 'https': proxy} def download(self, url, headers=None): # ... 在session.get中增加proxies参数 proxy = self.get_proxy() response = self.session.get(url, headers=final_headers, timeout=10, proxies=proxy) # ...

请求频率控制与随机延迟：我们已经实现了基础版本。更高级的可以针对不同域名设置不同的延迟策略。

5.3 扩展数据管道与调度策略

更多存储后端：继承Pipeline基类，可以轻松实现MongoDBPipeline、MySQLPipeline、JsonPipeline等，将数据存入不同目的地。
优先级调度：修改Scheduler，使用heapq实现优先队列。可以为不同的 URL 设置优先级（例如，详情页优先级高于列表页）。
断点续爬：将Scheduler中的seen_urls集合和任务队列定期序列化到磁盘（如用pickle保存）。爬虫重启时加载回来，可以从上次中断的地方继续。
布隆过滤器去重：当 URL 数量达到百万级以上时，内存中的set会非常庞大。可以使用pybloom-live等库实现布隆过滤器，用很小的内存和一定的误判率，实现海量 URL 去重。