网络爬虫架构解析：从控制节点到资源库的高效实现

1次阅读

没有评论

共计 3037 个字符，预计需要花费 8 分钟才能阅读完成。

一个完整的网络爬虫系统通常由三个核心组件构成，它们各司其职又紧密配合：

控制节点（Controller）：相当于系统的大脑，负责分配任务、调度爬虫节点、监控运行状态。它会维护待抓取 URL 队列，并根据策略将任务分发给爬虫节点。
爬虫节点（Crawler）：实际执行网页抓取工作的 ” 工人 ”，从控制节点获取任务，下载网页内容后，一方面将数据推送到资源库，另一方面解析出新发现的 URL 反馈给控制节点。
资源库（Repository）：存储抓取结果的数据库系统，通常需要支持高吞吐写入。根据业务需求可能采用关系型数据库、Elasticsearch 或对象存储等不同方案。

这三者通过消息队列（如 RabbitMQ/Kafka）进行解耦，形成典型的生产者 - 消费者模型。控制节点是任务生产者，爬虫节点是消费者，而资源库则是最终的目的地。

实际构建爬虫系统时，开发者常会遇到几个棘手问题：

高并发瓶颈 ：当需要同时抓取数千个网站时，单机爬虫的带宽和计算能力很快会成为瓶颈。更糟糕的是，某些网站会限制单个 IP 的访问频率。
数据去重 ：同一个 URL 可能被不同爬虫节点多次发现，如何避免重复抓取？特别是在分布式环境下，去重需要跨节点同步。
反爬对抗 ：现代网站普遍采用验证码、请求频率检测、User-Agent 验证等手段阻止爬虫。过于激进的抓取策略可能导致 IP 被封禁。
数据一致性 ：当爬虫节点意外崩溃时，如何确保任务不丢失？如何避免多个节点重复处理同一个 URL？

控制节点使用 Redis 的有序集合（ZSET）管理待抓取 URL，每个 URL 附带优先级分数。爬虫节点通过 BRPOP 命令获取任务，实现高效的分布式任务分配。

# 控制节点添加任务示例
redis.zadd('pending_urls', {'https://example.com/page1': 1, 'https://example.com/page2': 2})

# 爬虫节点获取任务示例
url = redis.brpop('pending_urls', timeout=30)

采用 RedisBloom 模块的布隆过滤器，在 O(1) 时间复杂度内判断 URL 是否已处理。虽然存在误判可能，但可以通过调整参数将误差率控制在 0.1% 以下。

from redisbloom.client import Client
rb = Client()

# 首次运行时创建过滤器
rb.bfCreate('visited_urls', 0.001, 1000000)

# 检查 URL 是否已存在
if not rb.bfExists('visited_urls', url):
    process_url(url)
    rb.bfAdd('visited_urls', url)

维护一个代理 IP 池，结合健康检查机制自动剔除失效的代理。建议使用异步 IO 库（如 aiohttp）配合多路复用，实现高效的轮询切换。

import random

class ProxyPool:
    def __init__(self):
        self.proxies = ['http://proxy1:port', 'http://proxy2:port']
        self.bad_proxies = set()

    def get_proxy(self):
        available = [p for p in self.proxies if p not in self.bad_proxies]
        return random.choice(available) if available else None

每个爬虫节点定期将处理状态保存到检查点（checkpoint），当节点重启时可以从最近的成功点继续，避免重复工作。推荐使用 SQLite 存储检查点数据。

以下是一个精简但功能完整的爬虫节点实现：

import aiohttp
from bs4 import BeautifulSoup
import asyncio
import redis

class CrawlerNode:
    def __init__(self):
        self.redis = redis.Redis(host='controller')
        self.session = aiohttp.ClientSession()

    async def fetch_page(self, url):
        try:
            async with self.session.get(url) as response:
                if response.status == 200:
                    return await response.text()
        except Exception as e:
            print(f"Failed to fetch {url}: {str(e)}")

    def parse_links(self, html):
        soup = BeautifulSoup(html, 'html.parser')
        return [a['href'] for a in soup.find_all('a', href=True)]

    async def run(self):
        while True:
            url = self.redis.brpop('pending_urls', timeout=30)
            if not url:
                continue

            html = await self.fetch_page(url)
            if html:
                # 存储到资源库
                self.redis.lpush('raw_data', html)

                # 提取新链接并提交
                new_links = self.parse_links(html)
                for link in new_links:
                    if not self.redis.bf_exists('visited_urls', link):
                        self.redis.zadd('pending_urls', {link: 1})

if __name__ == '__main__':
    crawler = CrawlerNode()
    asyncio.run(crawler.run())

我们对三种常见方案进行了基准测试（处理 100 万个 URL）：