Scrapy 分布式与生产部署

# Python 3.10+
# 优先级队列（默认）—— 基于 Redis Sorted Set，支持请求优先级调度
SCHEDULER_QUEUE_CLASS = 'scrapy_redis.queue.PriorityQueue'

# 先进先出队列 —— 基于 Redis List（lpush + rpop），广度优先
SCHEDULER_QUEUE_CLASS = 'scrapy_redis.queue.FifoQueue'

# 后进先出队列 —— 基于 Redis List（lpush + lpop），深度优先
SCHEDULER_QUEUE_CLASS = 'scrapy_redis.queue.LifoQueue'

# Python 3.10+
def request_seen(self, request):
    """判断 Request 是否重复"""
    fp = self.request_fingerprint(request)  # 计算 SHA1 指纹
    added = self.server.sadd(self.key, fp)  # 向 Redis Set 添加指纹
    return added == 0  # added=0 表示已存在（重复），added=1 表示新增

# Python 3.10+
# === settings.py ===
# 核心：替换调度器和去重过滤器
SCHEDULER = "scrapy_redis.scheduler.Scheduler"
DUPEFILTER_CLASS = "scrapy_redis.dupefilter.RFPDupeFilter"

# Redis 连接（推荐使用 URL 方式）
REDIS_URL = 'redis://:your_password@your_redis_host:6379'

# 持久化：爬取结束后不清空队列和去重集合
SCHEDULER_PERSIST = True

# 注意：分布式环境下不要设置 SCHEDULER_FLUSH_ON_START = True
# 否则每台 Worker 启动时都会清空队列，导致数据丢失

# 统一存储目标
MONGO_URI = 'mongodb://user:password@your_mongo_host:27017'

# 确保每台主机的 Python 环境和依赖一致
scrapy crawl your_spider_name

pip3 install scrapy-redis-bloomfilter

# Python 3.10+
# 使用 Bloom Filter 版本的去重过滤器
DUPEFILTER_CLASS = "scrapy_redis_bloomfilter.dupefilter.RFPDupeFilter"

# 位数组大小参数，默认 30（128MB，适合 1 亿级去重）
BLOOMFILTER_BIT = 30

# 哈希函数个数，默认 6
BLOOMFILTER_HASH_NUMBER = 6

# Python 3.10+
class BloomFilter:
    """基于 Redis Bitmap 的布隆过滤器"""
    def __init__(self, server, key, bit=30, hash_number=6):
        self.m = 1 << bit  # 位数组大小 = 2^bit
        self.server = server
        self.key = key
        # 用不同 seed 生成 k 个哈希函数
        self.hash_funcs = [HashMap(self.m, seed) for seed in range(hash_number)]

    def exists(self, value):
        """判断元素是否可能存在（True = 可能存在，False = 一定不存在）"""
        if not value:
            return False
        # 对所有哈希位置做 getbit，全部为 1 才返回 True
        return all(
            self.server.getbit(self.key, h.hash(value))
            for h in self.hash_funcs
        )

    def insert(self, value):
        """将元素插入过滤器"""
        for h in self.hash_funcs:
            self.server.setbit(self.key, h.hash(value), 1)

pip3 install scrapyd
scrapyd  # 默认监听 0.0.0.0:6800

[deploy:vm1]
url = http://120.27.34.24:6800/
project = weibo

[deploy:vm2]
url = http://139.217.26.30:6800/
project = weibo

# 部署到指定主机
scrapyd-deploy vm1

# 部署到所有已配置主机
scrapyd-deploy -a

FROM python:3.10
ADD . /code
WORKDIR /code
COPY ./scrapyd.conf /etc/scrapyd/
EXPOSE 6800
RUN pip install -r requirements.txt
CMD scrapyd

[scrapyd]
bind_address = 0.0.0.0   # 允许外部访问（容器内必须这样设置）
max_proc_per_cpu = 10    # 单核最多运行的爬虫任务数

# 构建镜像
docker build -t myproject/scrapyd:latest .

# 推送到 Docker Hub
docker tag myproject/scrapyd:latest registry.example.com/myproject/scrapyd:latest
docker push registry.example.com/myproject/scrapyd:latest

# 各主机拉取并运行
docker run -d --restart=always -p 6800:6800 registry.example.com/myproject/scrapyd

# Python 3.10+ 环境
# scrapyd_deploy.yml
---
- name: 批量部署 Scrapyd Docker 环境
  hosts: scrapyd_servers
  become: yes
  tasks:
    - name: 安装 Docker
      apt:
        name: docker.io
        state: present
        update_cache: yes

    - name: 拉取 Scrapyd 镜像
      docker_image:
        name: registry.example.com/myproject/scrapyd
        source: pull

    - name: 运行 Scrapyd 容器
      docker_container:
        name: scrapyd
        image: registry.example.com/myproject/scrapyd
        ports:
          - "6800:6800"
        restart_policy: always
        state: started

    - name: 验证 Scrapyd 服务
      uri:
        url: "http://{{ inventory_hostname }}:6800/daemonstatus.json"
        return_content: yes

[scrapyd_servers]
120.27.34.24  ansible_user=root
139.217.26.30 ansible_user=root
120.27.34.26  ansible_user=root

ansible-playbook -i hosts scrapyd_deploy.yml

pip3 install gerapy
gerapy init          # 初始化项目目录
gerapy migrate       # 初始化 SQLite 数据库
gerapy runserver     # 启动 Web 服务（默认 :8000）

# Python 3.10+
import redis
import requests
from datetime import datetime, timedelta

def check_workers_health(scrapyd_urls: list[str], redis_url: str) -> dict:
    """
    检查所有 Worker 和 Redis 的健康状态
    :return: 健康报告字典
    """
    report = {"timestamp": datetime.now().isoformat(), "workers": {}, "redis": {}}

    # 检查各 Worker
    for url in scrapyd_urls:
        try:
            resp = requests.get(f"{url}/daemonstatus.json", timeout=5)
            data = resp.json()
            report["workers"][url] = {
                "status": "ok" if data.get("status") == "ok" else "error",
                "running": data.get("running", 0),
                "pending": data.get("pending", 0),
                "finished": data.get("finished", 0),
            }
        except Exception as e:
            report["workers"][url] = {"status": "unreachable", "error": str(e)}

    # 检查 Redis
    r = redis.from_url(redis_url)
    info = r.info("memory")
    report["redis"] = {
        "used_memory_human": info.get("used_memory_human"),
        "connected_clients": info.get("connected_clients"),
        "uptime_in_days": info.get("uptime_in_days"),
    }

    return report