python 分布式爬虫、爬虫合规与综合实战项目

文章目录

• • 1 分布式爬虫原理
  • • 为什么需要分布式爬虫
    • 分布式爬虫架构
  • 2 Scrapy-Redis分布式框架
  • • 安装
    • 将普通Scrapy爬虫改造为分布式
    • 配置文件
    • 运行分布式爬虫
    • 断点续爬
  • 3 并发爬虫进阶
  • • asyncio + aiohttp 异步爬虫
  • 4 爬虫合规规范（必读）
  • • robots.txt规范
    • 合法爬虫边界
    • 近年相关法规参考
  • 5 综合实战项目
  • • 项目一：豆瓣电影数据分析系统
    • 项目二：电商价格监控系统
    • 项目三：定时数据采集脚本
  • 6 学习路线总结

1 分布式爬虫原理

为什么需要分布式爬虫

单机爬虫的瓶颈很明显：一台服务器的带宽、IP数量、计算能力都有限。当需要爬取海量数据时，分布式爬虫通过多台机器协同工作来突破这些限制。

适用场景：

• 爬取任务量大（百万级URL以上）
• 需要多IP轮换（被封IP后快速切换）
• 对实时性要求高（新闻监控、价格监控）
• 单机速度不够（24小时无法完成的任务）

分布式爬虫架构

Master节点（任务分发）
│
├── Redis任务队列（存储待爬URL）
│ │
├───────┼─────────────────┐
│ │ │
Worker1 Worker2 Worker3
(爬取) (爬取) (爬取)
│ │ │
└───────┴─────────────────┘
│
共享Redis去重队列
│
数据库存储

2 Scrapy-Redis分布式框架

Scrapy-Redis是最流行的分布式爬虫方案，基于Redis实现任务队列和去重。

安装

pip install scrapy-redis

需要运行Redis服务：

# Docker方式（推荐）
docker run -d -p 6379:6379 redis

# 或直接安装Redis
# Windows: 下载Redis for Windows
# Mac: brew install redis && redis-server
# Linux: apt-get install redis-server

将普通Scrapy爬虫改造为分布式

原来的普通爬虫：

import scrapy

class ProductSpider(scrapy.Spider):
name = 'product'
start_urls = ['https://example.com/products']

def parse(self, response):
pass

改造为分布式：

from scrapy_redis.spiders import RedisSpider

class ProductSpider(RedisSpider):
name = 'product'
redis_key = 'product:start_urls' # Redis中存储起始URL的key

def parse(self, response):
# 提取商品数据
for item in response.css('div.product'):
yield {
'title': item.css('h3::text').get(),
'price': item.css('span.price::text').get(),
'url': response.url
}

# 提取翻页链接（自动去重，不会重复爬取）
next_page = response.css('a.next::attr(href)').get()
if next_page:
yield response.follow(next_page, callback=self.parse)

配置文件

在 settings.py 中添加分布式配置：

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

# 使用Scrapy-Redis的去重
DUPEFILTER_CLASS = 'scrapy_redis.dupefilter.RFPDupeFilter'

# Redis连接
REDIS_URL = 'redis://localhost:6379'

# 不清空Redis队列（支持断点续爬）
SCHEDULER_PERSIST = True

# 并发请求数
CONCURRENT_REQUESTS = 32
CONCURRENT_REQUESTS_PER_DOMAIN = 16

# 下载延时
DOWNLOAD_DELAY = 0.5

运行分布式爬虫

启动所有Worker节点（每台机器执行）：

# 机器1
scrapy crawl product

# 机器2（相同命令，连接同一个Redis）
scrapy crawl product

# 机器3
scrapy crawl product

在Master节点推送起始URL：

import redis

r = redis.Redis(host='master-server-ip', port=6379)

# 推送起始URL
urls = [
'https://example.com/products?category=1',
'https://example.com/products?category=2',
'https://example.com/products?category=3'
]

for url in urls:
r.lpush('product:start_urls', url)

print(f'已推送 {len(urls)} 个URL到任务队列')

断点续爬

由于 SCHEDULER_PERSIST = True，爬虫意外中断后重启会从Redis中继续未完成的任务。

# 查看Redis中剩余任务数
import redis
r = redis.Redis(host='localhost', port=6379)
print(f'待爬URL数: {r.llen("product:start_urls")}')
print(f'已去重URL数: {r.scard("product:dupefilter")}')

3 并发爬虫进阶

asyncio + aiohttp 异步爬虫

对于不需要Scrapy的轻量场景，asyncio+aiohttp是高效的选择。

pip install aiohttp

import asyncio
import aiohttp
from bs4 import BeautifulSoup
import json
from datetime import datetime

async def fetch(session, url):
"""异步获取页面"""
headers = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64)'
}
try:
async with session.get(url, headers=headers, timeout=aiohttp.ClientTimeout(total=10)) as response:
if response.status == 200:
return await response.text()
return None
except Exception as e:
print(f'请求失败 {url}: {e}')
return None

async def parse_page(html, url):
"""解析页面（同步操作）"""
if not html:
return []

soup = BeautifulSoup(html, 'lxml')
items = []
for div in soup.select('div.product'):
items.append({
'title': div.select_one('h3').text.strip() if div.select_one('h3') else '',
'price': div.select_one('span.price').text.strip() if div.select_one('span.price') else '',
'url': url
})
return items

async def crawl_all(urls, concurrency=10):
"""并发爬取多个URL"""
semaphore = asyncio.Semaphore(concurrency) # 限制并发数
all_items = []

async def fetch_with_limit(session, url):
async with semaphore:
html = await fetch(session, url)
items = await asyncio.get_event_loop().run_in_executor(
None, parse_page, html, url # 在线程池中执行同步解析
)
return items

connector = aiohttp.TCPConnector(limit=concurrency)
async with aiohttp.ClientSession(connector=connector) as session:
tasks = [fetch_with_limit(session, url) for url in urls]
results = await asyncio.gather(*tasks)

for items in results:
all_items.extend(items)

return all_items

# 使用
urls = [f'https://example.com/products?page={i}' for i in range(1, 51)]
items = asyncio.run(crawl_all(urls, concurrency=10))
print(f'共爬取 {len(items)} 条数据')

4 爬虫合规规范（必读）

在学习爬虫技术的同时，必须了解法律边界和合规要求。这不是可选项。

robots.txt规范

每个网站都可以通过 robots.txt 文件声明爬虫规则。

# 示例 robots.txt
User-agent: * # 适用于所有爬虫
Disallow: /admin/ # 禁止爬取/admin/目录
Disallow: /user/ # 禁止爬取用户数据
Crawl-delay: 1 # 建议爬取间隔1秒

User-agent: Googlebot # 专门针对Google爬虫
Allow: / # 允许爬取所有内容

用Python检查robots.txt：

from urllib.robotparser import RobotFileParser

def can_fetch(url, user_agent='*'):
"""检查是否允许爬取该URL"""
from urllib.parse import urlparse
parsed = urlparse(url)
robots_url = f'{parsed.scheme}://{parsed.netloc}/robots.txt'

rp = RobotFileParser()
rp.set_url(robots_url)
try:
rp.read()
return rp.can_fetch(user_agent, url)
except:
return True # 无法读取则默认允许

# 使用
url = 'https://www.example.com/products'
if can_fetch(url):
print('允许爬取')
else:
print('robots.txt禁止爬取该URL')

Scrapy中默认遵守robots.txt（需确认 settings.py 中 ROBOTSTXT_OBEY = True）。

合法爬虫边界

可以爬取：

• 公开展示的新闻、博客、商品信息
• 统计、研究目的的公开数据
• 网站明确授权的数据

禁止爬取：

• 用户隐私数据（手机号、身份证、地址）
• 需要付费才能查看的内容
• 账号密码等安全信息
• 网站明确禁止的内容（robots.txt标注的）

行为准则：

• 控制请求频率，不攻击服务器
• 不爬取后用于商业牟利侵权
• 发现数据泄露风险及时停止

近年相关法规参考

• 《数据安全法》：对数据处理活动进行规范
• 《个人信息保护法》：保护个人信息不被滥用
• 《网络安全法》：保护网络基础设施安全

爬虫本身不违法，但爬取隐私数据、攻击服务器、侵权商用等行为可能触犯法律。

5 综合实战项目

项目一：豆瓣电影数据分析系统

功能：爬取豆瓣电影Top250数据，进行统计分析和可视化。

# 数据采集部分（完整版）
import requests
from bs4 import BeautifulSoup
import pandas as pd
import time
import random

def crawl_douban_movies():
movies = []
base_url = 'https://movie.douban.com/top250'

headers = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'
}

for start in range(0, 250, 25):
url = f'{base_url}?start={start}'
response = requests.get(url, headers=headers, timeout=10)
soup = BeautifulSoup(response.text, 'lxml')

for item in soup.select('div.item'):
movies.append({
'rank': item.select_one('em').text,
'title': item.select_one('span.title').text,
'rating': item.select_one('span.rating_num').text,
'votes': item.select_one('span[property="v:votes"]').text,
'year': item.select_one('span.inq').text.strip() if item.select_one('span.inq') else ''
})

time.sleep(random.uniform(1, 2))

return pd.DataFrame(movies)

# 数据分析部分
def analyze_movies(df):
df['rating'] = pd.to_numeric(df['rating'])
df['votes'] = pd.to_numeric(df['votes'])

print(f'平均评分: {df["rating"].mean():.2f}')
print(f'最高评分: {df["rating"].max()} – {df.loc[df["rating"].idxmax(), "title"]}')
print(f'最多投票: {df["votes"].max()} – {df.loc[df["votes"].idxmax(), "title"]}')

# 评分区间分布
bins = [8.0, 8.5, 9.0, 9.5, 10.0]
labels = ['8.0-8.5', '8.5-9.0', '9.0-9.5', '9.5+']
df['rating_range'] = pd.cut(df['rating'], bins=bins, labels=labels)
print('\\n评分分布:')
print(df['rating_range'].value_counts())

return df

# 运行
# df = crawl_douban_movies()
# df.to_csv('movies.csv', encoding='utf-8-sig', index=False)
# analyze_movies(df)

项目二：电商价格监控系统

import requests
import json
from datetime import datetime
import sqlite3

class PriceMonitor:
def __init__(self, db_path='prices.db'):
self.conn = sqlite3.connect(db_path)
self._init_db()

def _init_db(self):
"""初始化数据库"""
self.conn.execute('''
CREATE TABLE IF NOT EXISTS price_history (
id INTEGER PRIMARY KEY AUTOINCREMENT,
product_id TEXT,
product_name TEXT,
price REAL,
original_price REAL,
discount REAL,
record_time TEXT
)
''')
self.conn.commit()

def fetch_price(self, product_id):
"""获取商品价格（需根据具体网站调整）"""
headers = {'User-Agent': 'Mozilla/5.0 …'}
url = f'https://api.example.com/product/{product_id}'
response = requests.get(url, headers=headers, timeout=10)
data = response.json()

return {
'product_id': product_id,
'name': data.get('name', ''),
'price': data.get('current_price', 0),
'original_price': data.get('original_price', 0),
}

def record_price(self, product_data):
"""记录价格到数据库"""
price = product_data['price']
original = product_data['original_price']
discount = round((1 – price / original) * 100, 1) if original > 0 else 0

self.conn.execute('''
INSERT INTO price_history (product_id, product_name, price, original_price, discount, record_time)
VALUES (?, ?, ?, ?, ?, ?)
''', (
product_data['product_id'],
product_data['name'],
price, original, discount,
datetime.now().isoformat()
))
self.conn.commit()

def check_price_drop(self, product_id, threshold=10):
"""检查价格是否大幅下降"""
cursor = self.conn.execute('''
SELECT price FROM price_history
WHERE product_id = ?
ORDER BY record_time DESC
LIMIT 10
''', (product_id,))

prices = [row[0] for row in cursor.fetchall()]
if len(prices) < 2:
return False

latest = prices[0]
avg_recent = sum(prices[1:]) / len(prices[1:])
drop_percent = (avg_recent – latest) / avg_recent * 100

if drop_percent >= threshold:
print(f'价格预警！商品 {product_id} 价格下降 {drop_percent:.1f}%')
return True
return False

项目三：定时数据采集脚本

import schedule
import time

def run_crawler():
"""定时爬取任务"""
print(f'[{datetime.now()}] 开始爬取…')
# 执行爬虫逻辑
# …
print(f'[{datetime.now()}] 爬取完成')

# 每天早上9点运行
schedule.every().day.at("09:00").do(run_crawler)

# 每小时运行一次
schedule.every().hour.do(run_crawler)

# 每30分钟运行一次
schedule.every(30).minutes.do(run_crawler)

while True:
schedule.run_pending()
time.sleep(60) # 每分钟检查一次

Linux服务器部署定时任务（crontab）：

# 编辑crontab
crontab -e

# 每天早上8点运行爬虫
0 8 * * * /usr/bin/python3 /home/user/spider/main.py >> /home/user/spider/logs/cron.log 2>&1

# 查看任务列表
crontab -l

6 学习路线总结

学完这套教程，你应该掌握了：

基础层

• Python语法、数据结构、文件操作
• HTTP协议、网络请求、浏览器抓包
• HTML/CSS/XPath数据定位

技术层

• requests/urllib 发起请求
• BeautifulSoup/lxml/json 解析数据
• Selenium/Playwright 处理动态页面
• 接口逆向、参数分析

进阶层

• 反爬对抗：UA池、代理池、验证码识别
• Scrapy工程化框架
• Scrapy-Redis分布式爬虫
• asyncio并发爬虫

实战层

• 完整项目开发流程
• 数据库存储（MySQL/MongoDB）
• 定时任务与服务器部署
• 数据分析与可视化

持续提升方向：

爬虫是工具，数据是目的。学好爬虫技术的同时，要持续学习数据分析、可视化、机器学习，才能真正发挥爬取数据的价值。