1. IP代理池概念

IP代理池是指一组动态使用的代理服务器IP地址集合。通过使用代理池，用户可以在发送HTTP/HTTPS请求时随机或轮询不同的代理IP，以达到绕过限制、匿名访问或负载均衡的目的。代理池广泛应用于网络爬虫、数据采集、网络安全测试、反屏蔽等场景。

代理池的适用场景：

• 反封锁和绕过策略： 许多网站会检测访问请i去，尤其是对同一IP地址的连续请求，并可能会采取封禁IP、限制访问速率。使用代理池可以避免因为IP被封锁二而导致无法访问。
• 匿名和隐私保护： 使用代理池时，访问目标网站的请求并非来自用户的真实 IP，而是来自代理服务器。这样可以隐藏用户的真实 IP，提高匿名性。
• 提高并发性和效率 ： 通过使用多个代理IP，可以同时发送多个请求。在执行网络爬虫代码时非常必要重要，可以显著提高爬虫任务的效率。

2.IP代理池的工作原理

1. 获取代理IP： 可选方式
   • 从公开的代理网站抓取免费代理
   • 使用付费代理商的API提供的代理IP
   • 自己搭建代理服务器，或使用云服务器作为代理
2. 验证代理IP
   • 检查代理IP的可用性，通常会测试代理IP是否可正常访问目标网站。
   • 验证匿名性，确保代理IP不会泄露用户的真实IP。
3. 管理和使用代理IP
   • 构建代理IP池，将验证通过的IP地址加入池中，并定期检测IP的可用性。
   • 通过轮询、随机验证或根据策略分配代理IP来发送请求。
   • 当代理IP被封禁或失效时，自动从代理池中移除。

3.实战

构建IP代理池本次采用了一个github开源代码，仓库地址为仓库地址，要使用这个仓库代码构建代理池，需要安装redis，项目代码使用redis进行代理IP信息存储。本次利用这个开源代码实现http/https代理无感切换。

这个IP代理池默认搜集了以下网站的免费代理IP：

• 站大爷
• 66代理
• 开心代理
• FreeProxyList
• 快代理
• FateZero
• 云代理
• 小幻代理
• 免费代理
• 89代理
• 稻壳代理

3.1代理池构建

下载仓库代码之后，首先执行pip -r requeirement.txt安装所需依赖。

pip install  -i https://mirrors.aliyun.com/pypi/simple/ --trusted-host mirrors.aliyun.com     -r requirements.txt

安装依赖之后，修改配置文件setting.py配置文件，参照示例修改DD_CONNredis服务地址，修改HOST和PORT修改服务监听ip和端口。接下来执行如下命令启动代理池服务。

# 启动后台收集进程
python proxyPool.py schedule
# 启动提供resful接口进程
python proxyPool.py server

启动完成之后可以通过以下接口进行查询和获取到搜集到代理ip进行请求代理。

4.动态切换代理ip

无感动态切换代理ip，这次采用自定义实现request.Session的子类进行，在自定义子类中定义切换代理IP的规则，代码如下。

# encoding:utf-8
__author__ = 'ObsessedCE'
import requests
import time
import sys

"""
    自定义session，实现定义更换代理
"""

class RotatingProxySession(requests.Session):
    def __init__(self, proxy_pool_url="http://localhost:5010", switch_interval=60 * 10, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.proxy_pool_url = proxy_pool_url
        self.switch_interval = switch_interval
        self.last_switch_time = 0

    def _update_proxy(self):
        # 切换到下一个代理并更新代理设置
        new_proxy = self.__get_proxy_address()
        self.proxies = {
            "https": f"https://{new_proxy}",
            "http": f"http://{new_proxy}",
        }
        self.last_switch_time = time.time()
        print(f"Switched to proxy: {new_proxy}")

    def _check_and_update_proxy(self):
        # 检查时间是否到达切换间隔
        if time.time() - self.last_switch_time >= self.switch_interval:
            self._update_proxy()

    def request(self, *args, **kwargs):
        # 在每次请求前检查是否需要切换代理
        self._check_and_update_proxy()
        return super().request(*args, **kwargs)

    def __get_proxy_address(self):
        url = f"{self.proxy_pool_url}/get/?type=http"
        response = requests.get(url)
        if response.status_code != 200:
            print("获取代理地址失败")
            print(response.status_code)
            sys.exit(0)
        response_dict = response.json()
        proxy_url = response_dict.get("proxy")
        if not proxy_url:
            print("获取代理地址失败")
            sys.exit(0)
        return proxy_url

if __name__ == "__main__":
    session = RotatingProxySession(switch_interval=10)
    while True:
        # 获取请求ip
        response = session.get("http://httpbin.org/ip", timeout = 100)
        if response.status_code == 200:
            print(response.content)
        time.sleep(10)

执行效果: