当前位置：首页 > news >正文

实用指南：告别IP被封！分布式爬虫的“隐身”与“分身”术

news 2025/9/22 17:11:01

实用指南：告别IP被封！分布式爬虫的“隐身”与“分身”术

咱们平时上网爬材料，最头疼的就是IP被封。单台机器猛刷，网站一眼就能识破。想把活儿干得又快又稳，就得把任务拆开，让多台机器或多个进程一起干，每个还用不同的IP出口——这就好比让一群人轮流换装去排队，既减轻压力又降低风险。

在这里插入图片描述

下面是一个基于Python的分布式爬虫实现，启用多进程和代理隧道技术来分散请求压力并降低IP被封风险。

import multiprocessing
import requests
from urllib.parse import urlparse
import time
import random
from queue import Queue
import logging
from datetime import datetime
# 设置日志
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)
class DistributedCrawler
:
def __init__(self, proxy_list, num_processes=4, requests_per_ip=100, delay_range=(1, 3)):
"""
初始化分布式爬虫
Args:
proxy_list: 代理服务器列表，格式为 [{"http": "http://ip:port", "https": "https://ip:port"}, ...]
num_processes: 进程数量
requests_per_ip: 每个IP发出的请求数量限制
delay_range: 请求延迟范围(秒)
"""
self.proxy_list = proxy_list
self.num_processes = num_processes
self.requests_per_ip = requests_per_ip
self.delay_range = delay_range
self.task_queue = multiprocessing.Queue()
self.result_queue = multiprocessing.Queue()
self.proxy_usage = multiprocessing.Manager().dict()
# 初始化代理使用计数
for proxy in proxy_list:
key = self._get_proxy_key(proxy)
self.proxy_usage[key] = 0
def _get_proxy_key(self, proxy):
"""生成代理的唯一标识"""
http_proxy = proxy.get('http', '')
https_proxy = proxy.get('https', '')
return f"{http_proxy
}_{https_proxy
}"
def add_tasks(self, urls):
"""添加任务到队列"""
for url in urls:
self.task_queue.put(url)
# 添加结束信号
for _ in range(self.num_processes):
self.task_queue.put(None)
def get_available_proxy(self):
"""获取可用的代理，基于使用次数进行负载均衡"""
# 按使用次数排序，选择使用最少的代理
sorted_proxies = sorted(
self.proxy_list,
key=lambda p: self.proxy_usage[self._get_proxy_key(p)]
)
# 返回使用次数未超限的代理
for proxy in sorted_proxies:
key = self._get_proxy_key(proxy)
if self.proxy_usage[key] < self.requests_per_ip:
self.proxy_usage[key] += 1
return proxy
# 如果所有代理都超限，随机选择一个并重置计数
proxy = random.choice(self.proxy_list)
key = self._get_proxy_key(proxy)
self.proxy_usage[key] = 1
return proxy
def worker(self, worker_id):
"""工作进程函数"""
logger.info(f"Worker {worker_id
} started")
while True:
url = self.task_queue.get()
if url is None: # 结束信号
break
try:
# 获取代理
proxy = self.get_available_proxy()
logger.info(f"Worker {worker_id
} using proxy: {proxy
}")
# 添加随机延迟
delay = random.uniform(*self.delay_range)
time.sleep(delay)
# 发送请求
headers = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36'
}
response = requests.get(
url,
proxies=proxy,
headers=headers,
timeout=30
)
# 处理响应
if response.status_code == 200:
result = {
'url': url,
'content': response.text[:1000], # 只取前1000个字符
'status': 'success',
'worker_id': worker_id,
'proxy': proxy
}
self.result_queue.put(result)
logger.info(f"Worker {worker_id
} successfully crawled {url
}")
else:
self.result_queue.put({
'url': url,
'status': f'error: status code {response.status_code
}',
'worker_id': worker_id,
'proxy': proxy
})
logger.warning(f"Worker {worker_id
} got status code {response.status_code
} for {url
}")
except Exception as e:
self.result_queue.put({
'url': url,
'status': f'error: {
str(e)
}',
'worker_id': worker_id,
'proxy': proxy
})
logger.error(f"Worker {worker_id
} error crawling {url
}: {
str(e)
}")
logger.info(f"Worker {worker_id
} finished")
def run(self):
"""启动爬虫"""
processes = []
# 启动工作进程
for i in range(self.num_processes):
p = multiprocessing.Process(target=self.worker, args=(i,))
p.start()
processes.append(p)
# 等待所有进程完成
for p in processes:
p.join()
# 收集结果
results = []
while not self.result_queue.empty():
results.append(self.result_queue.get())
return results
# 示例使用
if __name__ == "__main__":
# 代理列表示例 (实际使用时需要替换为真实的代理IP)
proxy_list = [
{
"http": "http://proxy1.example.com:8080", "https": "https://proxy1.example.com:8080"
},
{
"http": "http://proxy2.example.com:8080", "https": "https://proxy2.example.com:8080"
},
{
"http": "http://proxy3.example.com:8080", "https": "https://proxy3.example.com:8080"
},
{
"http": "http://proxy4.example.com:8080", "https": "https://proxy4.example.com:8080"
},
]
# 要爬取的URL列表
urls_to_crawl = [
"https://httpbin.org/ip",
"https://httpbin.org/user-agent",
"https://httpbin.org/headers",
"https://httpbin.org/get",
] * 5 # 重复几次以展示分布式效果
# 创建分布式爬虫实例
crawler = DistributedCrawler(
proxy_list=proxy_list,
num_processes=4, # 使用4个进程
requests_per_ip=3, # 每个代理IP最多使用3次
delay_range=(1, 2) # 请求间隔1-2秒
)
# 添加任务
crawler.add_tasks(urls_to_crawl)
# 运行爬虫并获取结果
results = crawler.run()
# 打印结果摘要
success_count = sum(1 for r in results if r['status'] == 'success')
print(f"爬取完成! 成功: {success_count
}, 失败: {
len(results) - success_count
}")
# 打印部分结果
for i, result in enumerate(results[:3]):
print(f"结果 {i+1
}:")
print(f" URL: {result['url']
}")
print(f" 状态: {result['status']
}")
print(f" 工作进程: {result.get('worker_id', 'N/A')
}")
print(f" 使用的代理: {result.get('proxy', 'N/A')
}")
if 'content' in result:
print(f" 内容预览: {result['content'][:100]
}...")
print()