1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

# settings.py
# 启用 Scrapy-Redis 调度器
SCHEDULER = "scrapy_redis.scheduler.Scheduler"

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

# 启用 Scrapy-Redis Pipeline
ITEM_PIPELINES = {
'scrapy_redis.pipelines.RedisPipeline': 300
}

# Redis 连接配置
REDIS_HOST = 'localhost'
REDIS_PORT = 6379

# 队列持久化
SCHEDULER_PERSIST = True

# 请求队列的键
SCHEDULER_QUEUE_KEY = '%(spider)s:requests'

# 去重集合的键
SCHEDULER_DUPEFILTER_KEY = '%(spider)s:dupefilter'

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15

# spider.py
from scrapy_redis.spiders import RedisSpider

class MySpider(RedisSpider):
name = 'myspider'
redis_key = 'myspider:start_urls'

def parse(self, response):
# 解析逻辑
for item in self.extract_items(response):
yield item

# 提取新的 URL
for url in response.css('a::attr(href)').getall():
yield response.follow(url, self.parse)

1
2
3
4
5

# 在多台机器上启动爬虫
scrapy crawl myspider

# 向 Redis 添加起始 URL
redis-cli lpush myspider:start_urls "http://example.com"

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

# celery_app.py
from celery import Celery

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

app.conf.update(
task_serializer='json',
result_serializer='json',
accept_content=['json'],
timezone='Asia/Shanghai',
enable_utc=True,
)

@app.task(bind=True, max_retries=3)
def crawl_task(self, url):
"""爬取任务"""
try:
process = CrawlerProcess(get_project_settings())
process.crawl(MySpider, start_urls=[url])
process.start()
except Exception as exc:
raise self.retry(exc=exc, countdown=60)

1
2
3
4
5

# 启动 Celery Worker
celery -A celery_app worker --loglevel=info

# 使用 Flower 监控
celery -A celery_app flower

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21

from pyspider.libs.base_handler import *

class Handler(BaseHandler):
crawl_config = {
'itag': 'v1',
}

@every(minutes=24 * 60)
def on_start(self):
self.crawl('http://example.com', callback=self.index_page)

@config(age=10 * 24 * 60 * 60)
def index_page(self, response):
for each in response.doc('a[href^="http"]').items():
self.crawl(each.attr.href, callback=self.detail_page)

def detail_page(self, response):
return {
"url": response.url,
"title": response.doc('title').text(),
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

import redis
import hashlib

class RedisDeduplicator:
def __init__(self, host='localhost', port=6379):
self.r = redis.Redis(host=host, port=port)
self.key = 'crawler:urls'

def url_hash(self, url):
"""计算 URL 哈希"""
return hashlib.md5(url.encode()).hexdigest()

def is_duplicate(self, url):
"""检查 URL 是否重复"""
url_hash = self.url_hash(url)
return self.r.sismember(self.key, url_hash)

def add_url(self, url):
"""添加 URL 到去重集合"""
url_hash = self.url_hash(url)
return self.r.sadd(self.key, url_hash)

def check_and_add(self, url):
"""检查并添加（原子操作）"""
url_hash = self.url_hash(url)
# 返回 1 表示新 URL，0 表示重复
return self.r.sadd(self.key, url_hash) == 1

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

# 安装：pip install scrapy-redis-bloomfilter
# settings.py
DUPEFILTER_CLASS = 'scrapy_redis_bloomfilter.dupefilter.RFPDupeFilter'

# BloomFilter 配置
BLOOMFILTER_HASH_NUMBER = 6 # 哈希函数个数
BLOOMFILTER_BIT = 30 # 位数组大小 2^30 (约1G)

# Redis 配置
REDIS_HOST = 'localhost'
REDIS_PORT = 6379

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37

import redis
import mmh3 # pip install mmh3

class BloomFilter:
def __init__(self, host='localhost', port=6379, bit_size=30, hash_count=6):
self.r = redis.Redis(host=host, port=port)
self.key = 'crawler:bloomfilter'
self.bit_size = bit_size
self.hash_count = hash_count
self.size = 1 << bit_size # 2^bit_size

def _hash(self, url):
"""生成多个哈希值"""
hashes = []
for i in range(self.hash_count):
hash_val = mmh3.hash(url, i) % self.size
hashes.append(hash_val)
return hashes

def add(self, url):
"""添加 URL"""
for hash_val in self._hash(url):
self.r.setbit(self.key, hash_val, 1)

def contains(self, url):
"""检查 URL 是否存在"""
for hash_val in self._hash(url):
if not self.r.getbit(self.key, hash_val):
return False
return True

def check_and_add(self, url):
"""检查并添加"""
if self.contains(url):
return False # 已存在（或误判）
self.add(url)
return True # 新 URL

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22

import redis

class PriorityQueue:
def __init__(self, host='localhost', port=6379):
self.r = redis.Redis(host=host, port=port)
self.key = 'crawler:priority_queue'

def push(self, url, priority=0):
"""添加任务，priority 越大优先级越高"""
self.r.zadd(self.key, {url: -priority}) # 负值使得大优先级排前

def pop(self):
"""取出最高优先级任务"""
result = self.r.zpopmin(self.key, 1)
if result:
url, score = result[0]
return url.decode()
return None

def size(self):
"""队列大小"""
return self.r.zcard(self.key)

1
2
3
4
5
6
7
8
9

queue = PriorityQueue()

# 添加不同优先级的任务
queue.push('http://example.com/important', priority=10)
queue.push('http://example.com/normal', priority=5)
queue.push('http://example.com/low', priority=1)

# 按优先级取出
print(queue.pop()) # http://example.com/important

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

import hashlib

class ShardedCrawler:
def __init__(self, shard_id, total_shards):
self.shard_id = shard_id
self.total_shards = total_shards

def belongs_to_me(self, url):
"""判断 URL 是否属于当前分片"""
url_hash = int(hashlib.md5(url.encode()).hexdigest(), 16)
shard = url_hash % self.total_shards
return shard == self.shard_id

def should_crawl(self, url):
"""是否应该爬取此 URL"""
return self.belongs_to_me(url)

1
2
3
4
5

# 启动 4 个爬虫节点，每个负责 1/4 的 URL
python crawler.py --shard-id 0 --total-shards 4
python crawler.py --shard-id 1 --total-shards 4
python crawler.py --shard-id 2 --total-shards 4
python crawler.py --shard-id 3 --total-shards 4

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31

import redis
import time

class LoadBalancer:
def __init__(self, node_id):
self.r = redis.Redis()
self.node_id = node_id
self.heartbeat_key = f'crawler:heartbeat:{node_id}'
self.load_key = 'crawler:node_loads'

def report_heartbeat(self, current_load):
"""上报心跳和当前负载"""
self.r.setex(self.heartbeat_key, 30, current_load)
self.r.zadd(self.load_key, {self.node_id: current_load})

def get_active_nodes(self):
"""获取活跃节点"""
nodes = []
for key in self.r.scan_iter('crawler:heartbeat:*'):
node_id = key.decode().split(':')[-1]
load = int(self.r.get(key) or 0)
nodes.append((node_id, load))
return nodes

def get_least_loaded_node(self):
"""获取负载最小的节点"""
result = self.r.zrange(self.load_key, 0, 0, withscores=True)
if result:
node_id, load = result[0]
return node_id.decode(), int(load)
return None, None

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

distributed_crawler/
├── scrapy.cfg
├── crawler/
│ ├── __init__.py
│ ├── settings.py
│ ├── spiders/
│ │ ├── __init__.py
│ │ └── example_spider.py
│ ├── items.py
│ └── pipelines.py
└── requirements.txt

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34

# Scrapy-Redis 配置
SCHEDULER = "scrapy_redis.scheduler.Scheduler"
DUPEFILTER_CLASS = "scrapy_redis_bloomfilter.dupefilter.RFPDupeFilter"

# BloomFilter 配置
BLOOMFILTER_HASH_NUMBER = 6
BLOOMFILTER_BIT = 30 # 支持约 1 亿 URL

# Redis 配置
REDIS_HOST = 'redis.example.com'
REDIS_PORT = 6379
REDIS_PARAMS = {'password': 'yourpassword', 'db': 0}

# 持久化队列
SCHEDULER_PERSIST = True

# Pipeline 配置
ITEM_PIPELINES = {
'crawler.pipelines.MongoDBPipeline': 300,
}

# MongoDB 配置
MONGODB_URI = 'mongodb://localhost:27017'
MONGODB_DATABASE = 'crawler'

# 并发配置
CONCURRENT_REQUESTS = 32
CONCURRENT_REQUESTS_PER_DOMAIN = 8
DOWNLOAD_DELAY = 0.5

# 自动限速
AUTOTHROTTLE_ENABLED = True
AUTOTHROTTLE_START_DELAY = 1
AUTOTHROTTLE_TARGET_CONCURRENCY = 32

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31

# spiders/example_spider.py
from scrapy_redis.spiders import RedisSpider
from crawler.items import NewsItem

class ExampleSpider(RedisSpider):
name = 'example'
redis_key = 'example:start_urls'

custom_settings = {
'DEPTH_LIMIT': 3, # 最大深度
}

def parse(self, response):
"""解析列表页"""
# 提取新闻链接
for url in response.css('a.news-link::attr(href)').getall():
yield response.follow(url, callback=self.parse_detail)

# 提取下一页
next_page = response.css('a.next-page::attr(href)').get()
if next_page:
yield response.follow(next_page, callback=self.parse)

def parse_detail(self, response):
"""解析详情页"""
item = NewsItem()
item['url'] = response.url
item['title'] = response.css('h1::text').get()
item['content'] = response.css('div.content::text').getall()
item['publish_time'] = response.css('span.time::text').get()
yield item

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32

# pipelines.py
import pymongo
from itemadapter import ItemAdapter

class MongoDBPipeline:
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('MONGODB_URI'),
mongo_db=crawler.settings.get('MONGODB_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):
collection = self.db[spider.name]
item_dict = ItemAdapter(item).asdict()
collection.update_one(
{'url': item_dict['url']},
{'$set': item_dict},
upsert=True
)
return item

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17

# 在机器 A 启动爬虫
scrapy crawl example

# 在机器 B 启动爬虫
scrapy crawl example

# 在机器 C 启动爬虫
scrapy crawl example

# 添加起始 URL（在任意机器）
redis-cli -h redis.example.com -a yourpassword lpush example:start_urls "http://example.com"
redis-cli -h redis.example.com -a yourpassword lpush example:start_urls "http://example.com/page1"
redis-cli -h redis.example.com -a yourpassword lpush example:start_urls "http://example.com/page2"

# 查看队列状态
redis-cli -h redis.example.com -a yourpassword llen example:requests
redis-cli -h redis.example.com -a yourpassword scard example:dupefilter

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40

# tasks.py
from celery import Celery
import requests
from bs4 import BeautifulSoup

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

@app.task(bind=True, max_retries=3)
def crawl_url(self, url):
"""爬取单个 URL"""
try:
response = requests.get(url, timeout=10)
soup = BeautifulSoup(response.text, 'html.parser')

# 提取数据
data = {
'url': url,
'title': soup.find('h1').text if soup.find('h1') else '',
'links': [a['href'] for a in soup.find_all('a', href=True)]
}

# 存储数据
save_to_database(data)

# 提交新的爬取任务
for link in data['links'][:10]: # 限制每页最多提取 10 个链接
crawl_url.apply_async(args=[link], countdown=2)

return data

except Exception as exc:
# 失败后 60 秒重试
raise self.retry(exc=exc, countdown=60)

def save_to_database(data):
"""保存数据到数据库"""
# 实现数据存储逻辑
pass

1
2
3
4
5
6
7
8
9

# 启动多个 Worker
celery -A tasks worker --loglevel=info --concurrency=10

# 提交任务
python -c "from tasks import crawl_url; crawl_url.delay('http://example.com')"

# 监控
celery -A tasks flower
# 访问 http://localhost:5555

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

┌──────────────┐
│ Master 节点 │
│ (Redis) │
│ ┌──────────┐ │
│ │ 任务队列 │ │
│ │ 去重集合 │ │
│ │ BloomFilter│ │
│ └──────────┘ │
└──────┬───────┘
│
┌───┴────┬────────┬────────┐
↓ ↓ ↓ ↓
┌─────┐ ┌─────┐ ┌─────┐ ┌─────┐
│Slave1│ │Slave2│ │Slave3│ │SlaveN│
│爬虫节点│ │爬虫节点│ │爬虫节点│ │爬虫节点│
└──┬──┘ └──┬──┘ └──┬──┘ └──┬──┘
│ │ │ │
└────────┴────────┴────────┘
↓
┌───────────────┐
│ MongoDB │
│ (数据存储) │
└───────────────┘

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56

# Scrapy-Redis 分布式配置
SCHEDULER = "scrapy_redis.scheduler.Scheduler"
DUPEFILTER_CLASS = "scrapy_redis_bloomfilter.dupefilter.RFPDupeFilter"

# BloomFilter 配置（适配亿级数据）
BLOOMFILTER_HASH_NUMBER = 7
BLOOMFILTER_BIT = 33 # 2^33，支持 10 亿+ URL

# Redis 主节点配置
REDIS_HOST = '192.168.1.100'
REDIS_PORT = 6379
REDIS_PARAMS = {
'password': 'your_redis_password',
'db': 0,
'socket_keepalive': True,
'max_connections': 200,
}

# 持久化配置（支持断点续爬）
SCHEDULER_PERSIST = True
SCHEDULER_QUEUE_KEY = '%(spider)s:requests'
SCHEDULER_DUPEFILTER_KEY = '%(spider)s:dupefilter'

# Pipeline 配置
ITEM_PIPELINES = {
'jd_crawler.pipelines.MongoDBPipeline': 300,
'jd_crawler.pipelines.ImageDownloadPipeline': 400,
}

# MongoDB 配置
MONGODB_URI = 'mongodb://192.168.1.101:27017'
MONGODB_DATABASE = 'jd_products'
MONGODB_COLLECTION = 'products'

# 并发和延迟配置
CONCURRENT_REQUESTS = 64 # 总并发数
CONCURRENT_REQUESTS_PER_DOMAIN = 16 # 单域名并发
DOWNLOAD_DELAY = 0.2 # 下载延迟
RANDOMIZE_DOWNLOAD_DELAY = True # 随机化延迟

# 自动限速（防止被封）
AUTOTHROTTLE_ENABLED = True
AUTOTHROTTLE_START_DELAY = 0.5
AUTOTHROTTLE_MAX_DELAY = 3
AUTOTHROTTLE_TARGET_CONCURRENCY = 16.0

# User-Agent 轮换
DOWNLOADER_MIDDLEWARES = {
'scrapy.downloadermiddlewares.useragent.UserAgentMiddleware': None,
'jd_crawler.middlewares.RandomUserAgentMiddleware': 400,
'jd_crawler.middlewares.ProxyMiddleware': 410,
}

# Cookies 配置
COOKIES_ENABLED = True
COOKIES_DEBUG = False

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90

from scrapy_redis.spiders import RedisSpider
from jd_crawler.items import ProductItem
import json
import re

class JDSpider(RedisSpider):
name = 'jd'
redis_key = 'jd:start_urls'

# 允许的域名
allowed_domains = ['jd.com', '3.cn']

custom_settings = {
'DEPTH_LIMIT': 5, # 最大爬取深度
}

def parse(self, response):
"""解析商品列表页"""
# 提取商品链接
product_links = response.css('div.gl-i-wrap div.p-img a::attr(href)').getall()

for link in product_links:
if not link.startswith('http'):
link = 'https:' + link
yield scrapy.Request(link, callback=self.parse_product_detail)

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

def parse_product_detail(self, response):
"""解析商品详情页"""
item = ProductItem()

# 基本信息
item['product_id'] = self.extract_product_id(response.url)
item['url'] = response.url
item['title'] = response.css('div.sku-name::text').get('').strip()
item['brand'] = response.css('ul#parameter-brand li a::text').get('')

# 价格信息（需要额外请求）
price_url = f'https://p.3.cn/prices/mgets?skuIds=J_{item["product_id"]}'
yield scrapy.Request(
price_url,
callback=self.parse_price,
meta={'item': item}
)

def parse_price(self, response):
"""解析价格信息"""
item = response.meta['item']

try:
data = json.loads(response.text)
if data:
item['price'] = data[0].get('p', '0')
item['market_price'] = data[0].get('m', '0')
except:
item['price'] = '0'
item['market_price'] = '0'

# 评论数据（需要额外请求）
comment_url = f'https://club.jd.com/comment/productPageComments.action?productId={item["product_id"]}&score=0&sortType=5&page=0&pageSize=10'
yield scrapy.Request(
comment_url,
callback=self.parse_comment,
meta={'item': item}
)

def parse_comment(self, response):
"""解析评论信息"""
item = response.meta['item']

try:
data = json.loads(response.text)
item['comment_count'] = data.get('productCommentSummary', {}).get('commentCount', 0)
item['good_rate'] = data.get('productCommentSummary', {}).get('goodRate', 0)
item['good_count'] = data.get('productCommentSummary', {}).get('goodCount', 0)
except:
item['comment_count'] = 0
item['good_rate'] = 0
item['good_count'] = 0

yield item

def extract_product_id(self, url):
"""从 URL 中提取商品 ID"""
match = re.search(r'/(\d+)\.html', url)
return match.group(1) if match else ''

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48

import pymongo
from itemadapter import ItemAdapter
from datetime import datetime

class MongoDBPipeline:
"""MongoDB 存储 Pipeline"""

def __init__(self, mongo_uri, mongo_db, mongo_collection):
self.mongo_uri = mongo_uri
self.mongo_db = mongo_db
self.mongo_collection = mongo_collection
self.client = None
self.db = None
self.collection = None

@classmethod
def from_crawler(cls, crawler):
return cls(
mongo_uri=crawler.settings.get('MONGODB_URI'),
mongo_db=crawler.settings.get('MONGODB_DATABASE'),
mongo_collection=crawler.settings.get('MONGODB_COLLECTION')
)

def open_spider(self, spider):
self.client = pymongo.MongoClient(self.mongo_uri)
self.db = self.client[self.mongo_db]
self.collection = self.db[self.mongo_collection]

# 创建索引
self.collection.create_index('product_id', unique=True)
self.collection.create_index('crawl_time')

def close_spider(self, spider):
self.client.close()

def process_item(self, item, spider):
item_dict = ItemAdapter(item).asdict()
item_dict['crawl_time'] = datetime.now()

# 使用 upsert 避免重复
self.collection.update_one(
{'product_id': item_dict['product_id']},
{'$set': item_dict},
upsert=True
)

spider.logger.info(f'Item saved: {item_dict["title"]}')
return item

1
2
3
4
5
6
7
8
9

# 在 Master 服务器上启动 Redis
redis-server /etc/redis/redis.conf

# 配置 Redis 持久化
# 在 redis.conf 中设置：
# save 900 1
# save 300 10
# save 60 10000
# appendonly yes

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15

# 在 Slave 1 服务器上
cd /path/to/jd_crawler
scrapy crawl jd

# 在 Slave 2 服务器上
cd /path/to/jd_crawler
scrapy crawl jd

# 在 Slave 3 服务器上
cd /path/to/jd_crawler
scrapy crawl jd

# 在 Slave 4 服务器上
cd /path/to/jd_crawler
scrapy crawl jd

1
2
3
4
5
6

# 批量添加图书分类起始 URL
redis-cli -h 192.168.1.100 -a your_redis_password << EOF
lpush jd:start_urls "https://list.jd.com/list.html?cat=1713,3258,3274"
lpush jd:start_urls "https://list.jd.com/list.html?cat=1713,3258,9435"
lpush jd:start_urls "https://list.jd.com/list.html?cat=1713,3258,9438"
EOF

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17

# monitor.py - 实时监控脚本
import redis
import time

r = redis.Redis(host='192.168.1.100', password='your_redis_password', decode_responses=True)

while True:
queue_size = r.llen('jd:requests')
dupefilter_size = r.scard('jd:dupefilter')

print(f'[{time.strftime("%Y-%m-%d %H:%M:%S")}]')
print(f' 待爬取队列: {queue_size:,}')
print(f' 已去重URL: {dupefilter_size:,}')
print(f' 预计剩余时间: {queue_size / (64 * 5) / 60:.1f} 分钟') # 假设每秒 5 个请求/节点
print('-' * 50)

time.sleep(60)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18

# 列表页优先级高，详情页优先级低
class JDSpider(RedisSpider):
def parse(self, response):
# 列表页链接，高优先级
for link in product_links:
yield scrapy.Request(
link,
callback=self.parse_product_detail,
priority=1 # 详情页低优先级
)

# 翻页链接，最高优先级
if next_page:
yield response.follow(
next_page,
callback=self.parse,
priority=10 # 列表页高优先级
)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

class MongoDBPipeline:
def process_item(self, item, spider):
# 检查是否需要更新
existing = self.collection.find_one({'product_id': item_dict['product_id']})

if existing:
# 计算价格变化
old_price = float(existing.get('price', 0))
new_price = float(item_dict.get('price', 0))

if abs(old_price - new_price) > 0.01:
# 价格有变化，记录历史
self.collection.update_one(
{'product_id': item_dict['product_id']},
{
'$set': item_dict,
'$push': {
'price_history': {
'price': new_price,
'time': datetime.now()
}
}
}
)
else:
# 新商品，直接插入
self.collection.insert_one(item_dict)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15

class AutoThrottleMiddleware:
"""根据服务器响应动态调整请求速度"""

def process_response(self, request, response, spider):
if response.status == 429: # Too Many Requests
# 检测到限速，降低速度
spider.crawler.engine.downloader.total_concurrency.value -= 5
spider.logger.warning('Detected rate limiting, reducing concurrency')
elif response.status == 200:
# 正常响应，可以适当提速
current = spider.crawler.engine.downloader.total_concurrency.value
if current < 64:
spider.crawler.engine.downloader.total_concurrency.value += 1

return response

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26

# 使用账号池轮换
class WeiboSpider(RedisSpider):
name = 'weibo'

def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.account_pool = AccountPool() # 账号池管理

def start_requests(self):
# 从账号池获取 cookie
cookies = self.account_pool.get_cookie()
for url in self.start_urls:
yield scrapy.Request(url, cookies=cookies)

def parse_user(self, response):
# 解析用户信息
user_item = UserItem()
user_item['uid'] = response.meta['uid']
user_item['nickname'] = response.css('div.name::text').get()
# ... 其他字段

yield user_item

# 爬取粉丝和关注
followers_url = f'https://weibo.com/ajax/friendships/followers?uid={uid}'
yield scrapy.Request(followers_url, callback=self.parse_followers)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

# settings.py
CONCURRENT_REQUESTS = 100 # 全局并发
CONCURRENT_REQUESTS_PER_DOMAIN = 16 # 单域名并发
CONCURRENT_REQUESTS_PER_IP = 16 # 单 IP 并发

# 下载延迟
DOWNLOAD_DELAY = 0.25
RANDOMIZE_DOWNLOAD_DELAY = True

# 启用自动限速
AUTOTHROTTLE_ENABLED = True
AUTOTHROTTLE_START_DELAY = 0.5
AUTOTHROTTLE_MAX_DELAY = 10
AUTOTHROTTLE_TARGET_CONCURRENCY = 100

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

# 增加 Redis 连接池大小
REDIS_PARAMS = {
'socket_keepalive': True,
'socket_keepalive_options': {
socket.TCP_KEEPIDLE: 60,
socket.TCP_KEEPINTVL: 10,
socket.TCP_KEEPCNT: 3,
},
'health_check_interval': 30,
'max_connections': 100,
}

1
2
3
4

# settings.py
RETRY_ENABLED = True
RETRY_TIMES = 3
RETRY_HTTP_CODES = [500, 502, 503, 504, 408, 429]

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31

import redis
import time
from datetime import datetime

class NodeMonitor:
def __init__(self):
self.r = redis.Redis()
self.node_id = socket.gethostname()

def register_node(self):
"""注册节点"""
self.r.setex(f'node:{self.node_id}', 60, datetime.now().isoformat())

def heartbeat(self):
"""定期发送心跳"""
while True:
try:
self.register_node()
time.sleep(30)
except Exception as e:
print(f"Heartbeat failed: {e}")
time.sleep(5)

def check_dead_nodes(self):
"""检查失效节点"""
for key in self.r.scan_iter('node:*'):
if not self.r.exists(key):
node_id = key.decode().split(':')[1]
print(f"Node {node_id} is dead")
# 重新分配该节点的任务
self.redistribute_tasks(node_id)

1
2
3
4
5
6
7

# settings.py
DOWNLOADER_MIDDLEWARES = {
'crawler.middlewares.ProxyMiddleware': 100,
}

# 代理池配置
PROXY_POOL_URL = 'http://proxy-pool.example.com/get'

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

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

class RandomUserAgentMiddleware(UserAgentMiddleware):
def __init__(self, user_agent=''):
self.user_agents = [
'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)...',
]

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

1
2
3
4
5
6
7

# settings.py
DOWNLOAD_DELAY = 1
AUTOTHROTTLE_ENABLED = True
AUTOTHROTTLE_TARGET_CONCURRENCY = 2.0

# 使用令牌桶限流
from scrapy.extensions.throttle import AutoThrottle

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17

from prometheus_client import Counter, Histogram, Gauge
import time

# 定义指标
requests_total = Counter('crawler_requests_total', 'Total requests', ['spider', 'status'])
request_duration = Histogram('crawler_request_duration_seconds', 'Request duration')
queue_size = Gauge('crawler_queue_size', 'Queue size', ['spider'])

class MetricsMiddleware:
def process_request(self, request, spider):
request.meta['start_time'] = time.time()

def process_response(self, request, response, spider):
duration = time.time() - request.meta.get('start_time', time.time())
request_duration.observe(duration)
requests_total.labels(spider=spider.name, status=response.status).inc()
return response