写爬虫最头疼的事之一,就是跑着跑着 IP 被封了。用单 IP 硬撑,频率稍微高一点目标站就给你返回 403 或者验证码,效率直线下降。解决办法很直接——搭一个本地代理池,多 IP 轮换着用。

下面分享一个我自己在用的方案,用 Python 实现一个简单但够用的高可用代理池。

1. 整体思路

代理池的核心逻辑就三件事:

  1. 拉取 IP:通过代理 IP 接口批量获取可用 IP
  2. 验证筛选:用之前先检测一下,把不能用的踢掉
  3. 轮换使用:请求时从池子里取 IP,失败了换一个,池子快空了自动补货

2. 完整代码

下面是一个完整可运行的代理池类,直接复制就能用。

import requests
import random
import time
import threading


class ProxyPool:
    """简易高可用代理池"""

    def __init__(self, api_url, min_pool_size=5):
        self.api_url = api_url
        self.min_pool_size = min_pool_size
        self.proxies = []
        self.lock = threading.Lock()
        # 初始化时拉一批 IP
        self._fetch_proxies()

    def _fetch_proxies(self):
        """从 API 接口拉取代理 IP"""
        try:
            resp = requests.get(self.api_url, timeout=10)
            lines = resp.text.strip().split("\n")
            new_proxies = [line.strip() for line in lines if line.strip()]
            with self.lock:
                self.proxies.extend(new_proxies)
            print(f"[代理池] 拉取到 {len(new_proxies)} 个 IP,池中共 {len(self.proxies)} 个")
        except Exception as e:
            print(f"[代理池] 拉取失败: {e}")

    def _check_proxy(self, proxy):
        """检测单个代理是否可用"""
        try:
            requests.get(
                "https://httpbin.org/ip",
                proxies={"http": f"http://{proxy}", "https": f"http://{proxy}"},
                timeout=5,
            )
            return True
        except:
            return False

    def get_proxy(self):
        """获取一个可用代理,池子不够时自动补充"""
        with self.lock:
            if len(self.proxies) < self.min_pool_size:
                self._fetch_proxies()
            if not self.proxies:
                return None
            # 随机选取一个
            proxy = random.choice(self.proxies)
        return proxy

    def remove_proxy(self, proxy):
        """移除失效代理"""
        with self.lock:
            if proxy in self.proxies:
                self.proxies.remove(proxy)
                print(f"[代理池] 移除失效 IP: {proxy},剩余 {len(self.proxies)} 个")

    def request_with_retry(self, url, max_retries=3, **kwargs):
        """带自动重试和 IP 轮换的请求方法"""
        for i in range(max_retries):
            proxy = self.get_proxy()
            if not proxy:
                print("[代理池] 没有可用代理了")
                return None
            try:
                resp = requests.get(
                    url,
                    proxies={"http": f"http://{proxy}", "https": f"http://{proxy}"},
                    timeout=10,
                    **kwargs,
                )
                if resp.status_code == 200:
                    return resp
                else:
                    self.remove_proxy(proxy)
            except:
                self.remove_proxy(proxy)
        return None

使用起来很简单:

# 替换成你自己的 API 提取地址
API_URL = "https://proxy.horocn.com/api/v2/proxies?order_id=你的订单号&num=10&format=text&line_separator=LF"

pool = ProxyPool(api_url=API_URL, min_pool_size=5)

# 爬取目标页面
resp = pool.request_with_retry("https://example.com/data")
if resp:
    print(resp.text)

3. 几个关键设计点

为什么用随机选取而不是顺序轮换? 顺序轮换在多线程场景下容易出问题,多个线程同时用同一个 IP,反而更容易触发封禁。随机选取能把请求分散开。

min_pool_size 的作用: 池子里 IP 数量低于这个阈值就自动触发 API 拉取,避免出现"池子空了才去拉"的尴尬——那时候你的请求已经失败了。

失败自动剔除: 请求失败或返回非 200 状态码时,直接把这个 IP 踢出池子。代理 IP 的可用性本来就有时效性,没必要反复重试一个坏 IP。

4. 代理 IP 从哪来

代码里的 API_URL 需要替换成实际的代理 IP 接口地址。蜻蜓代理(proxy.horocn.com)提供私密代理 API 提取服务,每次可批量提取 10~50 个 IP,适合用来做代理池的数据源。

使用前需要先把你的服务器 IP 添加到白名单,在后台设置就行。私密代理经典版包天 25 元起,更多套餐可以看购买页

如果不想自己维护代理池,也可以直接用隧道代理(dyn.horocn.com:50000),每个请求自动分配随机 IP,省去了池子管理的麻烦:

# 隧道代理用法,无需维护代理池
proxies = {
    "http": "http://用户名:密码@dyn.horocn.com:50000",
    "https": "http://用户名:密码@dyn.horocn.com:50000",
}
resp = requests.get("https://example.com", proxies=proxies)

代理池方案灵活可控,隧道代理方案省心省力,根据自己的场景选就行。有问题可以联系蜻蜓代理的客服,技术人员直接对接,沟通效率比较高。注册试用

转载请注明