Python 数据采集中实现代理 IP 自动轮换：一个轻量级代理池设计

import requests
import time
import random
import threading
from queue import Queue

class ProxyPool:
    """轻量级代理池，支持健康检查和随机轮换"""
    
    def __init__(self, proxy_list=None, check_interval=60):
        self.proxy_queue = Queue()
        self.lock = threading.Lock()
        self.healthy = set()
        self.check_interval = check_interval
        if proxy_list:
            for proxy in proxy_list:
                self.proxy_queue.put(proxy)
        # 启动后台健康检查线程
        self._start_health_check()
    
    def _check_proxy(self, proxy, test_url="http://httpbin.org/ip", timeout=5):
        """检测单个代理是否可用，返回布尔值"""
        try:
            proxies = {"http": proxy, "https": proxy}
            start = time.time()
            resp = requests.get(test_url, proxies=proxies, timeout=timeout)
            if resp.status_code == 200 and (time.time() - start) < 2:
                return True
        except:
            pass
        return False
    
    def _health_check_loop(self):
        """后台定期健康检查"""
        while True:
            time.sleep(self.check_interval)
            with self.lock:
                to_check = list(self.healthy) if self.healthy else list(self.proxy_queue.queue)
            for proxy in to_check:
                if self._check_proxy(proxy):
                    with self.lock:
                        self.healthy.add(proxy)
                else:
                    with self.lock:
                        self.healthy.discard(proxy)
                        # 也可以将失效代理放回队列末尾，此处简单丢弃
    
    def _start_health_check(self):
        t = threading.Thread(target=self._health_check_loop, daemon=True)
        t.start()
    
    def get_proxy(self, strategy="random"):
        """获取一个可用代理，支持 random 策略"""
        with self.lock:
            if strategy == "random":
                if self.healthy:
                    return random.choice(list(self.healthy))
                elif not self.proxy_queue.empty():
                    # 队列中还有未检测的代理，取出一个并检测
                    proxy = self.proxy_queue.get()
                    if self._check_proxy(proxy):
                        self.healthy.add(proxy)
                        return proxy
            return None

def fetch_with_proxy_pool(url, proxy_pool, max_retries=3):
    """使用代理池发送请求，失败自动重试"""
    for attempt in range(max_retries):
        proxy = proxy_pool.get_proxy()
        if not proxy:
            time.sleep(2)
            continue
        try:
            proxies = {"http": proxy, "https": proxy}
            resp = requests.get(url, proxies=proxies, timeout=10)
            if resp.status_code == 200:
                return resp.text
        except Exception:
            pass
        time.sleep(random.uniform(1, 3))
    raise Exception("All proxies failed")

# 示例用法：以某住宅代理服务商（例如辣椒HTTP）提供的代理列表作为初始池
# 实际使用时请替换为真实代理地址（示例地址仅作演示，具体接入方式参考官方文档：https://www.lajiaohttp.com?kwd=hyj-csdn）
proxy_list = [
    "http://user:pass@proxy1.example.com:8080",
    "http://user:pass@proxy2.example.com:8080",
    # 更多代理...
]

pool = ProxyPool(proxy_list)
result = fetch_with_proxy_pool("https://httpbin.org/ip", pool)
print(result)