哎,你有没有试过爬数据爬到一半,突然IP就被封了?或者下载速度慢得像蜗牛,一个页面要加载半分钟?我反正遇到过,真的能把人急死。后来我才明白,问题不在代码,而在网络请求的方式上——说白了,就是你得学会用代理IP,而且得是高质量的那种。
你可能会问,为啥非得用代理?直接用自己的IP不行吗?行啊,当然行,如果你不介意每隔十分钟换一次IP或者被目标站拉黑名单的话。大多数网站都有反爬策略,频繁请求同一地址,轻则限速,重则封禁。这时候,代理IP就像是你的“隐身斗篷”或者“多重身份”,让你看起来像是来自不同地方的不同用户。
但随便找几个免费代理塞进代码里?别,千万别。免费代理大多数不稳定、速度慢,甚至还有安全风险。你永远不知道背后是谁在记录你的数据。所以咱们今天聊的是“高质量代理IP”——稳定、快速、匿名度高,能真正帮你提升效率的那种。
我一般会找一些口碑还不错的服务商,比如快代理,他们家IP池大,覆盖地区也多,响应速度在行业里算是比较靠谱的。注意,我不是在打广告啊,纯粹是个人用过觉得省心。你注册之后一般能拿到API提取链接,每天能提取一定量的IP,放进项目里轮询调用。
好了,不啰嗦了,直接上干货。
第一,怎么在Python里使用代理IP?如果你用requests库,一句代码就能搞定:
import requests
proxies = {
"http": "http://10.10.1.10:3128",
"https": "http://10.10.1.10:1080",
}
response = requests.get("http://example.com", proxies=proxies)
但这只是单个代理,我们要的是多个IP轮换,降低被封风险。所以一般会先获取一批IP,比如5个,放列表里,每次随机选一个用。
假设你从快代理拿到了5个IP,格式大概是这样的:ip:port
。你可能会拿到一个API返回的JSON,像这样:
{
"code": 0,
"data": [
{"ip": "111.111.111.111", "port": 8000},
{"ip": "112.112.112.112", "port": 8080},
...
]
}
接下来我们写段代码,随机选一个IP发起请求:
import requests
import random
# 假设你已经拿到了5个IP,放在这个列表里
proxies_list = [
"http://111.111.111.111:8000",
"http://112.112.112.112:8080",
"http://113.113.113.113:8888",
"http://114.114.114.114:8118",
"http://115.115.115.115:8090"
]
url = "https://httpbin.org/ip" # 这个网址可以返回当前使用的IP
def test_proxy(proxy):
try:
response = requests.get(url, proxies={"http": proxy, "https": proxy}, timeout=5)
print("成功!当前IP是:", response.json()['origin'])
except Exception as e:
print("这个代理挂了:", proxy)
# 随机选一个试试
random_proxy = random.choice(proxies_list)
test_proxy(random_proxy)
这只是一个最简单的用法。真实场景下,你还要处理IP失效、自动切换、并发请求这些问题。
比如说,你可以在Scrapy框架里用代理。Scrapy效率高,适合大规模采集。在middlewares.py里加一个代理中间件:
import random
class RandomProxyMiddleware(object):
def __init__(self, proxy_list):
self.proxies = proxy_list
@classmethod
def from_crawler(cls, crawler):
return cls(proxy_list=crawler.settings.get('PROXY_LIST'))
def process_request(self, request, spider):
proxy = random.choice(self.proxies)
request.meta['proxy'] = proxy
接着在settings.py里配置:
PROXY_LIST = [
'http://111.111.111.111:8000',
'http://112.112.112.112:8080',
# ... 剩下3个
]
DOWNLOADER_MIDDLEWARES = {
'myproject.middlewares.RandomProxyMiddleware': 543,
}
这样每个请求都会随机换一个代理IP。
但你肯定会遇到代理失效的情况。这时候最好有个机制自动剔除无效IP。我一般会写一个简单的验证器,每隔一段时间检查代理是否还活着:
import concurrent.futures
def check_proxy(proxy):
try:
res = requests.get('http://httpbin.org/ip', proxies={'http': proxy, 'https': proxy}, timeout=5)
if res.status_code == 200:
return True
except:
pass
return False
# 用线程池并发验证5个代理
valid_proxies = []
with concurrent.futures.ThreadPoolExecutor() as executor:
results = executor.map(check_proxy, proxies_list)
for proxy, is_valid in zip(proxies_list, results):
if is_valid:
valid_proxies.append(proxy)
print("可用的代理IP:", valid_proxies)
这段代码跑完,valid_proxies
里就只剩下还能用的IP了。你可以在采集任务开始前先跑一遍验证,避免中途掉链子。
说到效率,光有代理还不够,你得并发请求。Python里面,concurrent.futures或者aiohttp都能帮你实现:
import aiohttp
import asyncio
async def fetch(session, url, proxy):
try:
async with session.get(url, proxy=proxy) as response:
return await response.text()
except Exception as e:
print(f"请求失败:{e}")
return None
async def main():
urls = ["https://example.com/page1", "https://example.com/page2"] # 假设你要抓的链接列表
proxies = proxies_list # 刚才那5个代理
async with aiohttp.ClientSession() as session:
tasks = []
for i, url in enumerate(urls):
proxy = proxies[i % len(proxies)] # 循环使用代理
task = fetch(session, url, proxy)
tasks.append(task)
results = await asyncio.gather(*tasks)
# 处理results...
asyncio.run(main())
用这种方式,几十个页面可以同时开抓,每个请求走不同的代理,速度直接起飞。
末尾再提一嘴,代理IP的类型也很重要。一般分透明、匿名和高匿。高匿代理最好,完全不透露原始IP,适合严肃的数据采集任务。快代理那边好像高匿IP占比挺高,用起来不太会返回X-Forwarded-For头,隐蔽性不错。
当然啦,除了技术层面,还要注意目标网站的Robots协议,别把人家的服务器搞崩了,尽量模拟正常用户的行为,加上随机间隔时间,User-Agent轮换等等……这些细节改天再聊。
好了,差不多就这些。代码片段都是实际可用的,你可以直接复制到项目里改改用。记住,高质量代理IP不是万能药,但绝对是提升采集效率的关键一环。用好它,数据获取就会顺利很多。