本文目录导读：
我来详细说明Python爬虫去重机制的设计方案：
基础去重方案
内存集合去重（适合小规模）
class SimpleDeduplicator:
def __init__(self):
self.seen = set()
def is_duplicate(self, item):
if item in self.seen:
return True
self.seen.add(item)
return False
def clear(self):
self.seen.clear()
哈希去重（节省内存）
import hashlib
class HashDeduplicator:
def __init__(self):
self.seen = set()
def _hash(self, url):
return hashlib.md5(url.encode()).hexdigest()
def is_duplicate(self, url):
url_hash = self._hash(url)
if url_hash in self.seen:
return True
self.seen.add(url_hash)
return False
使用Redis实现分布式去重
import redis
import hashlib
class RedisDeduplicator:
def __init__(self, redis_host='localhost', redis_port=6379, key='crawler:visited'):
self.client = redis.Redis(host=redis_host, port=redis_port)
self.key = key
def is_duplicate(self, url):
url_hash = hashlib.md5(url.encode()).hexdigest()
return self.client.sismember(self.key, url_hash)
def add(self, url):
url_hash = hashlib.md5(url.encode()).hexdigest()
self.client.sadd(self.key, url_hash)
def batch_add(self, urls):
hashes = [hashlib.md5(url.encode()).hexdigest() for url in urls]
self.client.sadd(self.key, *hashes)
Bloom Filter去重（高效内存使用）
from pybloom_live import BloomFilter
class BloomDeduplicator:
def __init__(self, capacity=100000, error_rate=0.001):
self.bloom = BloomFilter(capacity, error_rate)
def is_duplicate(self, url):
return url in self.bloom
def add(self, url):
self.bloom.add(url)
自定义Bloom Filter实现
import math
import hashlib
from bitarray import bitarray
class CustomBloomFilter:
def __init__(self, n, fpr):
"""
n: 预计元素数量
fpr: 期望误判率
"""
self.n = n
self.fpr = fpr
self.m = self._calc_m(n, fpr) # 位数组大小
self.k = self._calc_k(n, self.m) # 哈希函数数量
self.bit_array = bitarray(self.m)
self.bit_array.setall(0)
def _calc_m(self, n, p):
return int(-n * math.log(p) / (math.log(2) ** 2))
def _calc_k(self, n, m):
return int((m / n) * math.log(2))
def _get_hashes(self, item):
hashes = []
for i in range(self.k):
hash_val = hashlib.md5(f"{item}{i}".encode()).hexdigest()
hashes.append(int(hash_val, 16) % self.m)
return hashes
def add(self, item):
for hash_val in self._get_hashes(item):
self.bit_array[hash_val] = 1
def is_duplicate(self, item):
for hash_val in self._get_hashes(item):
if not self.bit_array[hash_val]:
return False
return True
综合去重方案
import re
from urllib.parse import urlparse, urlunparse
from collections import deque
class AdvancedDeduplicator:
def __init__(self,
method='bloom',
max_size=1000000,
redis_config=None):
self.method = method
self.method = method
self.fingerprints = set() if method == 'set' else None
if method == 'bloom':
self.deduplicator = BloomDeduplicator(max_size)
elif method == 'redis':
self.deduplicator = RedisDeduplicator(**redis_config)
elif method == 'hash':
self.deduplicator = HashDeduplicator()
else: # set
self.deduplicator = SimpleDeduplicator()
def normalize_url(self, url):
"""URL标准化处理"""
parsed = urlparse(url)
# 去除片段标识
clean_url = parsed._replace(fragment='')
# 标准化路径
path = re.sub(r'/$', '', clean_url.path)
clean_url = clean_url._replace(path=path)
# 标准化查询参数（排序）
if clean_url.query:
params = sorted(clean_url.query.split('&'))
clean_url = clean_url._replace(query='&'.join(params))
return urlunparse(clean_url)
def extract_fingerprint(self, url):
"""提取URL指纹"""
normalized_url = self.normalize_url(url)
return normalized_url
def is_duplicate(self, url):
fingerprint = self.extract_fingerprint(url)
return self.deduplicator.is_duplicate(fingerprint)
def add(self, url):
fingerprint = self.extract_fingerprint(url)
self.deduplicator.add(fingerprint)
内容去重（检测重复内容）
import hashlib
from simhash import Simhash
class ContentDeduplicator:
def __init__(self, threshold=3):
self.threshold = threshold # SimHash汉明距离阈值
self.simhashes = []
def simhash_distance(self, hash1, hash2):
"""计算SimHash汉明距离"""
x = hash1 ^ hash2
return bin(x).count('1')
def is_duplicate_content(self, content):
"""判断内容是否重复"""
current_hash = Simhash(content).value
for existing_hash in self.simhashes:
distance = self.simhash_distance(current_hash, existing_hash)
if distance <= self.threshold:
return True
self.simhashes.append(current_hash)
return False
完整爬虫去重示例
import requests
from urllib.parse import urljoin
from bs4 import BeautifulSoup
class Crawler:
def __init__(self):
self.deduplicator = AdvancedDeduplicator(method='bloom', max_size=1000000)
self.queue = deque()
self.content_dedup = ContentDeduplicator()
def extract_links(self, url, html):
"""提取页面链接"""
soup = BeautifulSoup(html, 'html.parser')
links = []
for a_tag in soup.find_all('a', href=True):
href = a_tag['href']
full_url = urljoin(url, href)
# 过滤非HTTP链接
if not full_url.startswith(('http://', 'https://')):
continue
# 去重检查
if not self.deduplicator.is_duplicate(full_url):
links.append(full_url)
return links
def crawl(self, start_url, max_pages=100):
"""开始爬取"""
self.queue.append(start_url)
crawled_count = 0
while self.queue and crawled_count < max_pages:
url = self.queue.popleft()
if self.deduplicator.is_duplicate(url):
continue
try:
response = requests.get(url, timeout=10)
response.raise_for_status()
# 内容去重
if self.content_dedup.is_duplicate_content(response.text):
print(f"重复内容跳过: {url}")
continue
# 标记为已访问
self.deduplicator.add(url)
crawled_count += 1
print(f"爬取: {url} ({crawled_count}/{max_pages})")
# 提取新链接
new_links = self.extract_links(url, response.text)
for link in new_links:
self.queue.append(link)
except Exception as e:
print(f"爬取失败 {url}: {str(e)}")
使用建议
场景选择
- 小规模爬虫（<1万URL）：使用内存Set
- 中等规模（1-100万）：使用Bloom Filter
- 大规模分布式（>100万）：使用Redis去重需求**：添加SimHash
优化策略
# 1. 多级去重
class MultiLayerDeduplicator:
def __init__(self):
self.bloom = BloomDeduplicator()
self.redis_set = RedisDeduplicator()
def is_duplicate(self, url):
# 先在Bloom Filter检查（快速）
if not self.bloom.is_duplicate(url):
self.bloom.add(url)
return False
# Bloom Filter认为重复时，再到Redis确认
return self.redis_set.is_duplicate(url)
# 2. 定时清理
class TimedCleanDeduplicator:
def __init__(self, clean_interval=3600):
self.deduplicator = AdvancedDeduplicator()
self.clean_interval = clean_interval
self.last_clean = time.time()
def check_and_clean(self):
if time.time() - self.last_clean > self.clean_interval:
self.deduplicator.deduplicator.clear()
self.last_clean = time.time()
选择合适的去重方案取决于你的爬虫规模、运行环境和性能要求，建议从小规模开始,根据实际需求逐步升级。
