Scrapy 分布式采集完全指南

官方文档：https://docs.scrapy.org/
最后更新：2026-03-05

目录

1. Scrapy 架构总览
2. Spider 类详解
3. Request 与 Response 对象
4. Item 与 ItemLoader
5. 下载器中间件
6. Spider 中间件
7. Item Pipeline
8. 重要 settings 配置项
9. 反反爬策略
10. Scrapy-Redis 分布式改造
11. 分布式最佳实践
12. Scrapyd 部署
13. 常见陷阱与注意事项

1. Scrapy 架构总览

1.1 核心组件

┌─────────────────────────────────────────────────────┐
│                    Scrapy Engine                    │
│  (核心调度器，协调所有组件之间的数据流)               │
└──────┬──────────┬──────────┬──────────┬─────────────┘
       │          │          │          │
  ┌────▼───┐ ┌───▼────┐ ┌───▼────┐ ┌───▼────┐
  │Scheduler│ │Downlo- │ │ Spider │ │  Item  │
  │ 调度器  │ │ ader   │ │ 爬虫   │ │Pipeline│
  │         │ │下载器  │ │        │ │        │
  └────┬────┘ └───┬────┘ └───┬────┘ └────────┘
       │          │          │
  ┌────▼──────────▼──────────▼────┐
  │         Middleware            │
  │   (下载器中间件 + Spider中间件)│
  └───────────────────────────────┘

1.2 数据流说明

1.3 各组件职责

Engine（引擎）

• 控制所有组件之间的数据流
• 处理动作触发，不可自定义（框架核心）

Scheduler（调度器）

• 接收 Request 并排队
• 按顺序返回 Request 给 Engine
• 默认使用优先级队列 + 内存去重

Downloader（下载器）

• 执行实际的 HTTP 请求
• 将响应返回给 Engine/Spider
• 基于 Twisted 异步框架实现

Spider（爬虫）

• 定义爬取逻辑和解析逻辑
• 产生 Request 和 Item
• 可定义多个 callback 方法

Item Pipeline（数据管道）

• 处理 Spider 产生的 Item
• 清洗、验证、持久化数据
• 多个 Pipeline 按优先级串联执行

Middleware（中间件）

• 下载器中间件：处理 Request/Response
• Spider 中间件：处理 Spider 的输入/输出

2. Spider 类详解

2.1 Spider（基础爬虫）

import scrapy

class MySpider(scrapy.Spider):
    name = "my_spider"
    allowed_domains = ["example.com"]
    start_urls = ["https://example.com"]
    custom_settings = {
        "DOWNLOAD_DELAY": 1,
    }

    def parse(self, response):
        # 解析响应，产生 Item 或新 Request
        yield {"title": response.css("h1::text").get()}

Spider 类属性

Spider 核心方法

start_requests() 方法签名（可重写）

def start_requests(self):
    for url in self.start_urls:
        yield scrapy.Request(url, dont_filter=True)

2.2 CrawlSpider（链接爬取爬虫）

CrawlSpider 扩展了 Spider，通过定义规则自动跟进链接，适合全站爬取。

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

class MyCrawlSpider(CrawlSpider):
    name = "my_crawl_spider"
    allowed_domains = ["example.com"]
    start_urls = ["https://example.com"]

    rules = (
        # 跟进列表页链接，不解析
        Rule(LinkExtractor(restrict_css=".pagination a"), follow=True),
        # 跟进详情页链接，并调用 parse_item 解析
        Rule(LinkExtractor(restrict_css=".item a"), callback="parse_item"),
    )

    def parse_item(self, response):
        yield {
            "url": response.url,
            "title": response.css("h1::text").get(),
        }

注意：CrawlSpider 不能重写 parse() 方法，因为 CrawlSpider 内部使用 parse() 实现规则处理逻辑。自定义回调方法必须使用其他名称。

2.3 Rule 参数详解

Rule(link_extractor, callback=None, cb_kwargs=None, follow=None, process_links=None, process_request=None, errback=None)

2.4 LinkExtractor 参数详解

from scrapy.linkextractors import LinkExtractor

LinkExtractor(
    allow=(),
    deny=(),
    allow_domains=(),
    deny_domains=(),
    restrict_xpaths=(),
    restrict_css=(),
    tags=("a", "area"),
    attrs=("href",),
    canonicalize=False,
    unique=True,
    process_value=None,
    deny_extensions=None,
    restrict_text=None,
    strip=True,
)

2.5 其他内置 Spider 类型

3. Request 与 Response 对象

3.1 Request 参数详解

scrapy.Request(
    url,
    callback=None,
    method="GET",
    headers=None,
    body=None,
    cookies=None,
    meta=None,
    encoding="utf-8",
    priority=0,
    dont_filter=False,
    errback=None,
    flags=None,
    cb_kwargs=None,
)

FormRequest（表单请求）

scrapy.FormRequest(
    url,
    formdata=None,   # dict，表单数据，自动编码为 application/x-www-form-urlencoded
    **kwargs,        # 同 Request 的其他参数
)

FormRequest.from_response() 可自动填充页面中的表单：

scrapy.FormRequest.from_response(
    response,
    formname=None,       # 按 name 属性定位表单
    formid=None,         # 按 id 属性定位表单
    formnumber=0,        # 按序号定位表单（从 0 开始）
    formdata=None,       # 覆盖/补充表单字段
    formxpath=None,      # 按 XPath 定位表单
    formcss=None,        # 按 CSS 选择器定位表单
    clickdata=None,      # 模拟点击的按钮
    dont_click=False,    # 不模拟点击，直接提交
    **kwargs,
)

Request.meta 常用键

3.2 Response 属性与方法

response.follow() 参数

response.follow(
    url,              # str 或 Selector，相对或绝对 URL
    callback=None,
    method="GET",
    headers=None,
    body=None,
    cookies=None,
    meta=None,
    encoding="utf-8",
    priority=0,
    dont_filter=False,
    errback=None,
    cb_kwargs=None,
    flags=None,
)

3.3 Selector 选择器

# CSS 选择器
response.css("div.title::text").get()          # 获取第一个，不存在返回 None
response.css("div.title::text").getall()       # 获取所有，返回列表
response.css("a::attr(href)").get()            # 获取属性值
response.css("div.title::text").get(default="") # 指定默认值

# XPath
response.xpath("//div[@class='title']/text()").get()
response.xpath("//a/@href").getall()

# re 正则提取
response.css("div::text").re(r"\d+")           # 返回所有匹配列表
response.css("div::text").re_first(r"\d+")     # 返回第一个匹配

4. Item 与 ItemLoader

4.1 Item 定义

import scrapy

class ArticleItem(scrapy.Item):
    title = scrapy.Field()
    url = scrapy.Field()
    author = scrapy.Field()
    publish_date = scrapy.Field()
    content = scrapy.Field()
    tags = scrapy.Field()

scrapy.Field() 参数

Field 本质是一个字典，可以传入任意关键字参数作为元数据，这些元数据由 ItemLoader 或 Pipeline 使用：

class ArticleItem(scrapy.Item):
    title = scrapy.Field(
        input_processor=MapCompose(str.strip),    # ItemLoader 输入处理器
        output_processor=TakeFirst(),             # ItemLoader 输出处理器
        serializer=str,                           # 序列化函数（自定义用途）
    )

dataclass 风格（Scrapy 2.2+）

from dataclasses import dataclass, field

@dataclass
class ArticleItem:
    title: str = ""
    url: str = ""
    tags: list = field(default_factory=list)

4.2 ItemLoader

ItemLoader 提供了填充 Item 的机制，支持输入/输出处理器对数据进行清洗。

from scrapy.loader import ItemLoader
from itemloaders.processors import TakeFirst, MapCompose, Join
import w3lib.html

class ArticleLoader(ItemLoader):
    default_item_class = ArticleItem
    default_output_processor = TakeFirst()   # 默认输出处理器：取第一个值

    title_in = MapCompose(w3lib.html.remove_tags, str.strip)   # title 输入处理器
    content_in = MapCompose(w3lib.html.remove_tags)
    tags_out = Identity()                    # tags 输出处理器：保留列表

def parse(self, response):
    loader = ArticleLoader(item=ArticleItem(), response=response)
    loader.add_css("title", "h1::text")
    loader.add_css("content", "div.content")
    loader.add_xpath("author", "//span[@class='author']/text()")
    loader.add_value("url", response.url)
    return loader.load_item()

ItemLoader 初始化参数

ItemLoader 方法

常用内置处理器

5. 下载器中间件

5.1 中间件接口

自定义下载器中间件需实现以下一个或多个方法：

class MyDownloaderMiddleware:

    @classmethod
    def from_crawler(cls, crawler):
        # 可访问 crawler.settings
        instance = cls()
        return instance

    def process_request(self, request, spider):
        """
        对每个 Request 调用。
        返回值：
        - None：继续处理（交给下一个中间件或下载器）
        - Response：直接返回，不再下载
        - Request：将新 Request 放入调度器
        - 抛出 IgnoreRequest：调用 errback 或丢弃
        """
        return None

    def process_response(self, request, response, spider):
        """
        对每个 Response 调用。
        返回值：
        - Response（可以是新的）：继续处理
        - Request：将新 Request 放入调度器
        - 抛出 IgnoreRequest：调用 errback 或丢弃
        """
        return response

    def process_exception(self, request, exception, spider):
        """
        当 process_request() 或下载器抛出异常时调用。
        返回值：
        - None：继续处理异常
        - Response：停止异常处理，返回响应
        - Request：将新 Request 放入调度器
        """
        pass

5.2 内置下载器中间件（按默认优先级排序）

RetryMiddleware 相关 settings

6. Spider 中间件

6.1 中间件接口

class MySpiderMiddleware:

    @classmethod
    def from_crawler(cls, crawler):
        return cls()

    def process_spider_input(self, response, spider):
        """
        Response 传递给 Spider 之前调用。
        返回值：
        - None：继续处理
        - 抛出异常：调用 process_spider_exception()
        """
        return None

    def process_spider_output(self, response, result, spider):
        """
        Spider 处理 Response 后，对其产生的 Item/Request 调用。
        必须返回 Item 和 Request 的可迭代对象。
        """
        for item in result:
            yield item

    def process_spider_exception(self, response, exception, spider):
        """
        Spider 或 process_spider_input() 抛出异常时调用。
        返回值：
        - None：继续传播异常
        - 可迭代对象：正常处理
        """
        pass

    def process_start_requests(self, start_requests, spider):
        """
        对 Spider 的初始请求调用，必须返回 Request 的可迭代对象。
        """
        for request in start_requests:
            yield request

6.2 内置 Spider 中间件

7. Item Pipeline

7.1 Pipeline 接口

class MyPipeline:

    @classmethod
    def from_crawler(cls, crawler):
        # 可从 settings 读取配置
        instance = cls()
        instance.db_url = crawler.settings.get("DATABASE_URL")
        return instance

    def open_spider(self, spider):
        """Spider 启动时调用，适合初始化资源（数据库连接等）"""
        self.conn = create_connection(self.db_url)

    def close_spider(self, spider):
        """Spider 关闭时调用，适合释放资源"""
        self.conn.close()

    def process_item(self, item, spider):
        """
        对每个 Item 调用，必须返回 Item 或抛出 DropItem。
        返回值：
        - Item：继续传给下一个 Pipeline
        - 抛出 DropItem：丢弃此 Item，不再传给后续 Pipeline
        """
        if not item.get("title"):
            raise DropItem(f"Missing title: {item}")
        self.conn.insert(item)
        return item

7.2 常见 Pipeline 实现

MongoDB Pipeline

import pymongo
from scrapy.exceptions import DropItem

class MongoPipeline:
    collection_name = "articles"

    @classmethod
    def from_crawler(cls, crawler):
        instance = cls()
        instance.mongo_uri = crawler.settings.get("MONGO_URI", "mongodb://localhost:27017")
        instance.mongo_db = crawler.settings.get("MONGO_DATABASE", "scrapy")
        return instance

    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[self.collection_name].insert_one(dict(item))
        return item

去重 Pipeline

from scrapy.exceptions import DropItem

class DuplicatesPipeline:
    def __init__(self):
        self.ids_seen = set()

    def process_item(self, item, spider):
        if item["url"] in self.ids_seen:
            raise DropItem(f"Duplicate item: {item['url']}")
        self.ids_seen.add(item["url"])
        return item

7.3 在 settings 中启用 Pipeline

ITEM_PIPELINES = {
    "myproject.pipelines.ValidationPipeline": 200,   # 数字越小越先执行
    "myproject.pipelines.DuplicatesPipeline": 300,
    "myproject.pipelines.MongoPipeline": 400,
}

8. 重要 settings 配置项

8.1 爬虫行为

8.2 并发与限速

8.3 下载器

8.4 重试与错误处理

8.5 去重

8.6 缓存

8.7 中间件与 Pipeline 启用

8.8 Feed 导出

FEEDS 配置示例

FEEDS = {
    "output/%(name)s_%(time)s.json": {
        "format": "json",          # 格式：json, jsonlines, csv, xml
        "encoding": "utf8",
        "store_empty": False,      # 是否在没有 Item 时也生成文件
        "item_classes": [ArticleItem],  # 只导出这些 Item 类型
        "fields": ["title", "url"],     # 只导出这些字段
        "indent": 4,               # JSON 缩进
        "overwrite": True,         # 是否覆盖已有文件
    },
}

8.9 日志

9. 反反爬策略

9.1 User-Agent 轮换

# middlewares.py
import random

class RandomUserAgentMiddleware:
    USER_AGENTS = [
        "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36",
        "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) 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 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/17.2 Safari/605.1.15",
    ]

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

# settings.py
DOWNLOADER_MIDDLEWARES = {
    "scrapy.downloadermiddlewares.useragent.UserAgentMiddleware": None,  # 禁用默认
    "myproject.middlewares.RandomUserAgentMiddleware": 400,
}

也可使用 fake-useragent 库：

from fake_useragent import UserAgent

class FakeUserAgentMiddleware:
    def __init__(self):
        self.ua = UserAgent()

    def process_request(self, request, spider):
        request.headers["User-Agent"] = self.ua.random

9.2 代理中间件

import random

class ProxyMiddleware:
    PROXIES = [
        "http://proxy1.example.com:8080",
        "http://proxy2.example.com:8080",
        "http://user:[email protected]:8080",
    ]

    def process_request(self, request, spider):
        request.meta["proxy"] = random.choice(self.PROXIES)

    def process_response(self, request, response, spider):
        # 代理失效时切换
        if response.status == 403:
            request.meta["proxy"] = random.choice(self.PROXIES)
            return request  # 重新发送请求
        return response

    def process_exception(self, request, exception, spider):
        # 连接失败时切换代理重试
        request.meta["proxy"] = random.choice(self.PROXIES)
        return request

9.3 Cookies 管理策略

# settings.py

# 方案一：禁用 cookies（避免 session 追踪）
COOKIES_ENABLED = False

# 方案二：按 Spider 隔离 cookies（每个 Spider 独立 cookie jar）
# 在 Request.meta 中指定 cookiejar 编号
def parse(self, response):
    yield Request(url, meta={"cookiejar": 1})  # 使用 cookie jar 1
    yield Request(url, meta={"cookiejar": 2})  # 使用独立的 cookie jar 2

# 方案三：手动传入已有 cookies
yield Request(url, cookies={"session": "abc123", "token": "xyz"})

COOKIES 相关配置

9.4 下载延迟策略

# settings.py

# 固定延迟
DOWNLOAD_DELAY = 2          # 每次请求等待 2 秒
RANDOMIZE_DOWNLOAD_DELAY = True  # 随机化为 1~3 秒

# 自动限速（推荐，根据服务器响应时间动态调整）
AUTOTHROTTLE_ENABLED = True
AUTOTHROTTLE_START_DELAY = 2
AUTOTHROTTLE_MAX_DELAY = 30
AUTOTHROTTLE_TARGET_CONCURRENCY = 2.0  # 目标同时向服务器发出 2 个并发请求
AUTOTHROTTLE_DEBUG = True

9.5 其他反反爬措施

# 设置完整的浏览器请求头
DEFAULT_REQUEST_HEADERS = {
    "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,*/*;q=0.8",
    "Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8",
    "Accept-Encoding": "gzip, deflate, br",
    "Connection": "keep-alive",
    "Upgrade-Insecure-Requests": "1",
}

# 处理 Cloudflare 等 JS 挑战（需配合 scrapy-playwright 或 scrapy-splash）
DOWNLOAD_HANDLERS = {
    "http": "scrapy_playwright.handler.ScrapyPlaywrightDownloadHandler",
    "https": "scrapy_playwright.handler.ScrapyPlaywrightDownloadHandler",
}

# 对特定请求启用 Playwright
yield Request(url, meta={"playwright": True})

9.6 Splash 渲染 JS 页面

# settings.py
SPLASH_URL = "http://localhost:8050"
DOWNLOADER_MIDDLEWARES = {
    "scrapy_splash.SplashCookiesMiddleware": 723,
    "scrapy_splash.SplashMiddleware": 725,
    "scrapy.downloadermiddlewares.httpcompression.HttpCompressionMiddleware": 810,
}
SPIDER_MIDDLEWARES = {
    "scrapy_splash.SplashDeduplicateArgsMiddleware": 100,
}
DUPEFILTER_CLASS = "scrapy_splash.SplashAwareDupeFilter"

# Spider 中使用
from scrapy_splash import SplashRequest

yield SplashRequest(url, callback=self.parse, args={"wait": 2})

10. Scrapy-Redis 分布式改造

10.1 分布式原理

Scrapy 默认使用内存作为请求队列，单机运行。Scrapy-Redis 将以下组件替换为 Redis 实现：

多个 Scrapy 节点连接同一 Redis，共享请求队列和去重集合，实现真正的分布式爬取。

┌─────────┐  ┌─────────┐  ┌─────────┐
│ Scrapy  │  │ Scrapy  │  │ Scrapy  │
│ Node 1  │  │ Node 2  │  │ Node 3  │
└────┬────┘  └────┬────┘  └────┬────┘
     │             │             │
     └─────────────┼─────────────┘
                   │ 共享
            ┌──────▼──────┐
            │    Redis     │
            │  请求队列    │
            │  去重集合    │
            │  起始URL     │
            └─────────────┘

10.2 安装

pip install scrapy-redis

10.3 基础配置

# settings.py

# Redis 连接
REDIS_URL = "redis://username:password@localhost:6379/0"
# 或分别配置
REDIS_HOST = "localhost"
REDIS_PORT = 6379
REDIS_DB = 0
REDIS_PASSWORD = None
REDIS_ENCODING = "utf-8"

# 使用 Scrapy-Redis 的 Scheduler
SCHEDULER = "scrapy_redis.scheduler.Scheduler"

# 使用 Redis 去重过滤器
DUPEFILTER_CLASS = "scrapy_redis.dupefilter.RFPDupeFilter"

# 是否在 Redis 中持久化请求队列（爬虫关闭后保留，支持断点续爬）
SCHEDULER_PERSIST = True

# 请求队列类型
SCHEDULER_QUEUE_CLASS = "scrapy_redis.queue.SpiderPriorityQueue"
# 可选值：
# "scrapy_redis.queue.SpiderQueue"         FIFO 队列（先进先出）
# "scrapy_redis.queue.SpiderStack"         LIFO 队列（后进先出/深度优先）
# "scrapy_redis.queue.SpiderPriorityQueue" 优先级队列（默认，推荐）

# 队列中无请求时的等待时间（秒）
SCHEDULER_IDLE_BEFORE_CLOSE = 0   # 0 表示立即关闭；设置 > 0 可等待新任务

# Redis key 前缀（所有 Scrapy-Redis key 都以此为命名空间）
REDIS_ITEMS_KEY = "%(spider)s:items"
REDIS_START_URLS_KEY = "%(name)s:start_urls"

# 是否将 Item 序列化到 Redis（用于后续处理）
REDIS_ITEMS_SERIALIZER = "json"

SCHEDULER_QUEUE_CLASS 对比

10.4 Spider 改造

将 scrapy.Spider 替换为 scrapy_redis.spiders.RedisSpider：

from scrapy_redis.spiders import RedisSpider

class MyDistributedSpider(RedisSpider):
    name = "my_distributed_spider"
    allowed_domains = ["example.com"]

    # Redis key，Spider 从此 key 读取起始 URL
    # 默认为 "%(name)s:start_urls"，即 "my_distributed_spider:start_urls"
    redis_key = "my_distributed_spider:start_urls"

    # 每次从 Redis 批量获取的 URL 数量（0 表示不限制）
    redis_batch_size = 16

    # 无任务时的最大等待时间（秒）
    max_idle_time = 0

    def parse(self, response):
        # 解析逻辑与普通 Spider 完全一致
        yield {
            "url": response.url,
            "title": response.css("h1::text").get(),
        }

向 Redis 推送起始 URL

# 使用 redis-cli 推送
redis-cli lpush my_distributed_spider:start_urls "https://example.com/page1"
redis-cli lpush my_distributed_spider:start_urls "https://example.com/page2"

# 批量推送
redis-cli rpush my_distributed_spider:start_urls \
    "https://example.com/page1" \
    "https://example.com/page2" \
    "https://example.com/page3"

用 Python 推送

import redis
import json

r = redis.Redis(host="localhost", port=6379, db=0)

urls = [
    "https://example.com/page1",
    "https://example.com/page2",
]

for url in urls:
    r.lpush("my_distributed_spider:start_urls", url)

# 推送带 meta 的完整 Request（序列化后推送）
request_data = {
    "url": "https://example.com/category/1",
    "callback": "parse_category",
    "meta": {"category": "tech"},
}
r.lpush("my_distributed_spider:start_urls", json.dumps(request_data))

10.5 RedisCrawlSpider（链接爬虫分布式版本）

from scrapy_redis.spiders import RedisCrawlSpider
from scrapy.spiders import Rule
from scrapy.linkextractors import LinkExtractor

class MyDistributedCrawlSpider(RedisCrawlSpider):
    name = "my_crawl_spider"
    allowed_domains = ["example.com"]
    redis_key = "my_crawl_spider:start_urls"

    rules = (
        Rule(LinkExtractor(allow=r"/list/"), follow=True),
        Rule(LinkExtractor(allow=r"/detail/"), callback="parse_item"),
    )

    def parse_item(self, response):
        yield {
            "url": response.url,
            "title": response.css("h1::text").get(),
        }

10.6 DUPEFILTER_CLASS 详解

RFPDupeFilter（Scrapy-Redis 默认去重）

DUPEFILTER_CLASS = "scrapy_redis.dupefilter.RFPDupeFilter"

工作原理：

1. 对每个 Request 计算指纹（fingerprint）：对 URL 规范化后与 method、body 组合进行 SHA1 哈希
2. 将指纹写入 Redis 集合（SADD spider:dupefilter fingerprint）
3. 通过 SISMEMBER 判断是否已处理

Redis 中的 key

自定义指纹算法

from scrapy_redis.dupefilter import RFPDupeFilter
from scrapy.utils.request import fingerprint

class CustomDupeFilter(RFPDupeFilter):
    def request_fingerprint(self, request):
        # 自定义指纹：只用 URL，忽略参数顺序
        from w3lib.url import canonicalize_url
        return hashlib.sha1(canonicalize_url(request.url).encode()).hexdigest()

10.7 Scheduler 详解

SCHEDULER = "scrapy_redis.scheduler.Scheduler"

Scheduler 相关配置

11. 分布式最佳实践

11.1 多节点部署方案

目录结构

my_scrapy_project/
├── my_scrapy_project/
│   ├── spiders/
│   │   └── distributed_spider.py
│   ├── middlewares.py
│   ├── pipelines.py
│   └── settings.py
├── scrapy.cfg
└── requirements.txt

settings.py 完整分布式配置

# ===== 基础配置 =====
BOT_NAME = "my_scrapy_project"
SPIDER_MODULES = ["my_scrapy_project.spiders"]
NEWSPIDER_MODULE = "my_scrapy_project.spiders"

# ===== Scrapy-Redis 分布式配置 =====
REDIS_URL = "redis://:[email protected]:6379/0"
SCHEDULER = "scrapy_redis.scheduler.Scheduler"
DUPEFILTER_CLASS = "scrapy_redis.dupefilter.RFPDupeFilter"
SCHEDULER_PERSIST = True               # 断点续爬
SCHEDULER_FLUSH_ON_START = False       # 启动时不清空队列
SCHEDULER_IDLE_BEFORE_CLOSE = 60      # 空闲 60 秒后关闭
SCHEDULER_QUEUE_CLASS = "scrapy_redis.queue.SpiderPriorityQueue"

# ===== 并发配置 =====
CONCURRENT_REQUESTS = 32
CONCURRENT_REQUESTS_PER_DOMAIN = 8
DOWNLOAD_DELAY = 0.5
RANDOMIZE_DOWNLOAD_DELAY = True
AUTOTHROTTLE_ENABLED = True
AUTOTHROTTLE_TARGET_CONCURRENCY = 4.0

# ===== 中间件 =====
DOWNLOADER_MIDDLEWARES = {
    "scrapy.downloadermiddlewares.useragent.UserAgentMiddleware": None,
    "my_scrapy_project.middlewares.RandomUserAgentMiddleware": 400,
    "my_scrapy_project.middlewares.ProxyMiddleware": 750,
}

# ===== Pipeline =====
ITEM_PIPELINES = {
    "my_scrapy_project.pipelines.ValidationPipeline": 200,
    "my_scrapy_project.pipelines.MongoPipeline": 400,
}

# ===== 日志 =====
LOG_LEVEL = "INFO"
LOG_FILE = "/var/log/scrapy/%(name)s.log"

在多台机器上启动 Spider

# 机器 1
scrapy crawl my_distributed_spider

# 机器 2（同一 Redis，不同节点，并行运行）
scrapy crawl my_distributed_spider

# 机器 3
scrapy crawl my_distributed_spider

所有节点从同一 Redis 队列取任务，结果写入同一 MongoDB，实现真正的分布式采集。

11.2 断点续爬

配置 SCHEDULER_PERSIST = True 后，爬虫关闭时 Redis 中的请求队列和去重集合会保留。

# 查看剩余待爬 URL 数量
redis-cli llen my_distributed_spider:requests

# 查看已去重 URL 数量
redis-cli scard my_distributed_spider:dupefilter

# 重新启动时，自动从上次中断处继续
scrapy crawl my_distributed_spider

全新开始（清空状态）

# 方法一：通过 settings
SCHEDULER_FLUSH_ON_START = True   # 启动时清空

# 方法二：手动清空 Redis
redis-cli del my_distributed_spider:requests
redis-cli del my_distributed_spider:dupefilter
redis-cli del my_distributed_spider:start_urls

11.3 URL 去重策略对比

Bloom Filter 去重（适合亿级 URL）

# 使用 scrapy-redis-bloomfilter
pip install scrapy-redis-bloomfilter

# settings.py
DUPEFILTER_CLASS = "scrapy_redis_bloomfilter.dupefilter.RFPDupeFilter"
BLOOMFILTER_HASH_NUMBER = 6     # hash 函数数量，越多误判率越低（内存越大）
BLOOMFILTER_BIT = 30            # 2^30 bit = 128MB，控制内存占用

11.4 生产环境监控

# 使用 Scrapy 内置的 stats
class MonitorPipeline:
    def open_spider(self, spider):
        self.start_time = time.time()

    def close_spider(self, spider):
        stats = spider.crawler.stats.get_stats()
        duration = time.time() - self.start_time
        spider.logger.info(f"采集完成: {stats.get('item_scraped_count', 0)} 条, 耗时 {duration:.1f}s")

# 通过 Scrapy 命令查看实时统计（Scrapyd 部署时）
curl http://localhost:6800/daemonstatus.json

# 通过 Redis 监控队列长度
redis-cli monitor   # 实时命令流
redis-cli info stats  # 统计信息

11.5 Item 后处理架构

在大规模分布式场景下，建议将 Item 写入 Redis 队列，由独立的消费者进程处理：

# Pipeline：将 Item 写入 Redis 队列
import redis
import json

class RedisQueuePipeline:
    @classmethod
    def from_crawler(cls, crawler):
        instance = cls()
        instance.redis_url = crawler.settings.get("REDIS_URL")
        instance.redis_key = crawler.settings.get("REDIS_ITEMS_KEY", "%(spider)s:items")
        return instance

    def open_spider(self, spider):
        self.r = redis.from_url(self.redis_url)
        self.key = self.redis_key % {"spider": spider.name}

    def process_item(self, item, spider):
        self.r.rpush(self.key, json.dumps(dict(item), ensure_ascii=False))
        return item

# 独立消费者进程
import redis
import json
import pymongo

r = redis.Redis(host="localhost", port=6379)
mongo = pymongo.MongoClient()["scrapy"]

while True:
    _, data = r.blpop("my_spider:items", timeout=30)
    if data:
        item = json.loads(data)
        mongo["articles"].insert_one(item)

12. Scrapyd 部署

12.1 安装与配置

pip install scrapyd scrapyd-client

# 启动 Scrapyd 服务（默认监听 6800 端口）
scrapyd

scrapyd.conf 配置文件

[scrapyd]
eggs_dir    = eggs          # 项目 egg 存储目录
logs_dir    = logs          # 日志目录
items_dir   =               # Item 存储目录（空则不存储）
jobs_to_keep = 5            # 每个 Spider 保留的历史任务数
dbs_dir     = dbs           # 数据库目录
max_proc    = 0             # 最大并行进程数，0 表示 CPU 数量
max_proc_per_cpu = 4        # 每个 CPU 的最大并行进程数
finished_to_keep = 100      # 保留的已完成任务数
poll_interval = 5.0         # 轮询间隔（秒）
bind_address = 127.0.0.1    # 监听地址（改为 0.0.0.0 对外开放）
http_port   = 6800          # 监听端口
debug       = off
runner      = scrapyd.runner
application = scrapyd.app.application
launcher    = scrapyd.launcher.Launcher
webroot     = scrapyd.website.Root

12.2 部署项目

# scrapy.cfg 配置
[settings]
default = my_project.settings

[deploy:production]
url = http://192.168.1.200:6800/
project = my_project

[deploy:staging]
url = http://192.168.1.201:6800/
project = my_project

# 部署到生产环境
scrapyd-deploy production -p my_project

# 查看已部署版本
curl http://192.168.1.200:6800/listversions.json?project=my_project

12.3 Scrapyd API

启动爬虫任务

curl http://localhost:6800/schedule.json \
    -d project=my_project \
    -d spider=my_distributed_spider \
    -d setting=DOWNLOAD_DELAY=2 \
    -d custom_arg=value

# 返回
{"status": "ok", "jobid": "6487ec79947edab326d6db28a2d86511e8247444"}

查看任务日志

# 日志路径：logs/{project}/{spider}/{jobid}.log
tail -f logs/my_project/my_distributed_spider/6487ec79947edab326d6db28a2d86511e8247444.log

12.4 多节点 Scrapyd 管理

对于大规模分布式部署，可使用 scrapydweb 进行可视化管理：

pip install scrapydweb
scrapydweb  # 默认 5000 端口

# 或使用 spiderkeeper
pip install spiderkeeper
spiderkeeper --server=http://localhost:6800

13. 常见陷阱与注意事项

13.1 CrawlSpider 的 parse 方法陷阱

# 错误：重写 parse 方法会破坏 CrawlSpider 的规则处理
class MyCrawlSpider(CrawlSpider):
    rules = (Rule(LinkExtractor(), callback="parse"),)  # 不要用 parse 作为 callback

    def parse(self, response):   # 不要重写 parse
        pass

# 正确：使用其他方法名
class MyCrawlSpider(CrawlSpider):
    rules = (Rule(LinkExtractor(), callback="parse_item"),)

    def parse_item(self, response):
        pass

13.2 Request 去重陷阱

# 陷阱：同 URL 的 POST 请求默认被去重
# 默认去重指纹包含 method 和 body，但 FormRequest 可能出现去重问题

# 解决：对特殊请求禁用去重
yield Request(url, dont_filter=True)

# 或自定义去重指纹
from scrapy.utils.request import fingerprint

class CustomDupeFilter(RFPDupeFilter):
    def request_fingerprint(self, request):
        # 在指纹计算中加入自定义维度
        base = fingerprint(request)
        custom = request.meta.get("custom_id", "")
        return hashlib.sha1((base + custom).encode()).hexdigest()

13.3 Spider 关闭信号陷阱

# 陷阱：在 close_spider 中访问 spider.crawler.stats 需要注意时序
class MyPipeline:
    def close_spider(self, spider):
        # 此时 stats 已最终化，可以安全访问
        count = spider.crawler.stats.get_value("item_scraped_count")
        spider.logger.info(f"共采集 {count} 条数据")

13.4 Item 与 dict 的区别

# Item 在访问不存在的键时抛出 KeyError（与 dict 一致）
item = ArticleItem()
item["nonexistent"]   # KeyError

# 但 Item 只允许访问已定义的字段
item["undefined_field"] = "value"  # KeyError，不允许赋值未定义字段

# 安全访问
item.get("title")            # 不存在返回 None
item.get("title", "default") # 指定默认值

# 转换为 dict
dict(item)

13.5 Scrapy-Redis 的 SCHEDULER_IDLE_BEFORE_CLOSE

# 陷阱：默认值为 0，队列一空爬虫立即退出
# 在分布式场景中，其他节点可能还在产生新 URL，设置等待时间避免过早退出
SCHEDULER_IDLE_BEFORE_CLOSE = 60   # 队列为空后等待 60 秒再关闭

13.6 内存泄漏风险

# 陷阱：在 Spider 类变量中积累数据导致内存泄漏
class MySpider(scrapy.Spider):
    seen_urls = []  # 类变量，所有实例共享，且不会被释放

    def parse(self, response):
        self.seen_urls.append(response.url)   # 持续增长，导致内存泄漏

# 正确：使用实例变量或依赖 Scrapy 的去重机制
class MySpider(scrapy.Spider):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.seen_urls = set()   # 实例变量，随 Spider 生命周期管理

13.7 Pipeline 中的异步操作

# 在 Pipeline 中需要异步操作时，使用 Twisted 或 asyncio
from twisted.internet.defer import inlineCallbacks
from scrapy.utils.defer import maybe_deferred

class AsyncMongoPipeline:
    @inlineCallbacks
    def process_item(self, item, spider):
        # 使用 motor（异步 MongoDB 驱动）
        yield self.collection.insert_one(dict(item))
        return item

# Scrapy 2.0+ 支持原生 asyncio
class AsyncPipeline:
    async def process_item(self, item, spider):
        await self.collection.insert_one(dict(item))
        return item

13.8 allowed_domains 与子域名

# 陷阱：allowed_domains 中的域名会匹配子域名
# "example.com" 不会匹配 "www.example.com"，需要明确列出

class MySpider(scrapy.Spider):
    allowed_domains = ["example.com", "www.example.com", "api.example.com"]

# 或者禁用 OffsiteMiddleware
SPIDER_MIDDLEWARES = {
    "scrapy.spidermiddlewares.offsite.OffsiteMiddleware": None,
}

13.9 cookies 与多账号

# 使用 cookiejar 隔离不同账号的 session
class MultiAccountSpider(scrapy.Spider):
    accounts = [
        {"username": "user1", "password": "pass1"},
        {"username": "user2", "password": "pass2"},
    ]

    def start_requests(self):
        for i, account in enumerate(self.accounts):
            yield scrapy.FormRequest(
                "https://example.com/login",
                formdata=account,
                meta={"cookiejar": i},    # 每个账号独立的 cookie jar
                callback=self.after_login,
                cb_kwargs={"account_id": i},
            )

    def after_login(self, response, account_id):
        yield scrapy.Request(
            "https://example.com/data",
            meta={"cookiejar": account_id},   # 继续使用同一 cookie jar
            callback=self.parse_data,
        )

13.10 分布式去重的一致性

Scrapy-Redis 使用 Redis SADD/SISMEMBER 实现去重，这些操作是原子的，但在极高并发下仍可能出现竞态条件（多个节点同时判断同一 URL 未爬取）。

# 对于幂等操作（如读取），重复爬取影响不大，可以接受
# 对于非幂等操作（如写入），需要在 Pipeline 层面做去重
class DatabasePipeline:
    def process_item(self, item, spider):
        # 使用 upsert 操作而非 insert，避免重复写入
        self.collection.update_one(
            {"url": item["url"]},          # 查询条件
            {"$set": dict(item)},           # 更新内容
            upsert=True,                    # 不存在则插入
        )
        return item

快速参考

项目初始化

# 创建项目
scrapy startproject my_project
cd my_project

# 创建 Spider
scrapy genspider my_spider example.com
scrapy genspider -t crawl my_crawl_spider example.com

# 运行
scrapy crawl my_spider
scrapy crawl my_spider -o output.json
scrapy crawl my_spider -s LOG_LEVEL=INFO

# 调试
scrapy shell "https://example.com"
scrapy shell --nolog "https://example.com"

# 检查 Spider 输出
scrapy parse "https://example.com" --spider=my_spider -d 2

常用 Shell 命令

# 在 scrapy shell 中
fetch("https://example.com")           # 重新获取 URL
response.css("h1::text").get()
response.xpath("//h1/text()").get()
response.json()
view(response)                          # 在浏览器中打开响应

分布式快速启动清单

# 1. 确认 Redis 可连接
redis-cli -h 192.168.1.100 ping

# 2. 推送起始 URL
redis-cli -h 192.168.1.100 rpush my_spider:start_urls "https://example.com"

# 3. 在所有节点启动 Spider
scrapy crawl my_spider

# 4. 监控队列
redis-cli -h 192.168.1.100 llen my_spider:requests    # 待爬数量
redis-cli -h 192.168.1.100 scard my_spider:dupefilter  # 已去重数量

# 5. 断点续爬（确认 SCHEDULER_PERSIST=True，直接重启即可）
scrapy crawl my_spider

最佳实践

将 Item 定义为 dataclass 或用 ItemLoader，不要在 Spider 里手动构建字典：Item 提供字段定义和校验；ItemLoader 支持输入/输出处理器链，适合复杂的字段清洗（去空格、拼接列表、类型转换），比分散的字符串处理更易维护。

通过 settings.py 控制并发和速率，而非多进程启动多个 Scrapy 进程：CONCURRENT_REQUESTS、DOWNLOAD_DELAY、AUTOTHROTTLE_ENABLED 可以精细控制爬取速率，避免对目标网站造成压力，也避免被封 IP。

为所有 Spider 设置 custom_settings，实现 Spider 级别的配置覆盖：不同 Spider 可能需要不同的并发数、User-Agent 或 Cookie 策略，通过 custom_settings 覆盖全局配置，无需启动多个项目。

使用 Pipeline 做数据持久化，Item 只做数据传递：Scrapy Pipeline 支持异步写入（MySQL、MongoDB、ES），与 Spider 解耦，便于切换存储后端。关闭 Pipeline 时不影响爬虫逻辑。

分布式爬取时使用 scrapy-redis + Redis 实现请求队列共享和去重：多台机器共用同一个 Redis 请求队列，RFPDupeFilter 确保 URL 不重复爬取，SCHEDULER_PERSIST=True 支持断点续爬。

常见陷阱

陷阱：在 parse 方法中使用同步阻塞 IO

现象： Spider 单线程爬取，即使设置了 CONCURRENT_REQUESTS=32，吞吐量也极低；或 Scrapy 控制台卡住不动。

原因： Scrapy 基于 Twisted 异步框架，parse 方法在 reactor 线程中执行，任何同步阻塞调用（time.sleep()、同步数据库写入、requests.get()）都会阻塞整个爬虫。

解决： 使用 scrapy-playwright/scrapy-splash 处理 JS 渲染，Pipeline 中用异步 DB 驱动（aiomysql、motor），不要在 Spider 中调用同步网络请求。

陷阱：ROBOTSTXT_OBEY=True 导致大量 URL 被跳过

现象： 爬虫启动后没有任何请求，或只处理了少量 URL，日志显示"Forbidden by robots.txt"。

原因： 默认配置 ROBOTSTXT_OBEY=True 会先获取并解析目标站点的 robots.txt，不允许的路径会被忽略。

解决： 在 settings.py 中设置 ROBOTSTXT_OBEY = False（合法合规前提下），或仔细检查 robots.txt 了解允许爬取的路径范围。

陷阱：重复爬取同一 URL（去重失效）

现象： 日志显示同一 URL 被请求多次，或数据库中出现大量重复数据。

原因： 在 parse 回调中用 yield Request(url) 时未检查 URL 是否已在队列或已访问，或自定义了 fingerprint 导致相同 URL 产生不同指纹。

解决： 确认 DUPEFILTER_CLASS 为默认的 RFPDupeFilter；自定义请求时不要随意修改 meta、headers（这些影响指纹），或显式传 dont_filter=False。

参见

• httpx完全指南
• BeautifulSoup与lxml完全指南
• Redis完全指南
• Playwright完全指南