深入理解 HTTP 代理：从协议原理到手写一个简易代理池

客户端 → 直接连接目标服务器 → 发送 HTTP 请求

客户端 → 连接代理服务器 → 发送完整URL的HTTP请求 → 代理服务器 → 连接目标服务器 → 转发请求

# 没有代理
GET /index.html HTTP/1.1
Host: example.com

# 有代理
GET http://example.com/index.html HTTP/1.1
Host: example.com

GET http://example.com/ HTTP/1.1
Proxy-Authorization: Basic dXNlcjpwYXNz

import time
import random
import requests
from dataclasses import dataclass
from typing import List

@dataclass
class Proxy:
    ip: str
    port: int
    protocol: str = "http"
    last_check: float = 0
    fail_count: int = 0
    success_count: int = 0

    @property
    def url(self) -> str:
        return f"{self.protocol}://{self.ip}:{self.port}"

    @property
    def success_rate(self) -> float:
        total = self.success_count + self.fail_count
        return self.success_count / total if total > 0 else 0

class ProxyPool:
    def __init__(self, check_url: str = "http://httpbin.org/ip"):
        self.proxies: List[Proxy] = []
        self.check_url = check_url
        self.blacklist = set()  # 失败次数过多的IP加入黑名单

    def add_proxy(self, proxy: Proxy):
        """添加代理IP"""
        if proxy.url not in self.blacklist:
            self.proxies.append(proxy)

    def get_proxy(self) -> Proxy | None:
        """获取一个可用代理（加权随机策略）"""
        available = [p for p in self.proxies if p.fail_count < 5]
        if not available:
            return None
        
        # 按成功率加权，成功率高的被选中概率大
        weights = [max(p.success_rate, 0.1) for p in available]
        return random.choices(available, weights=weights, k=1)[0]

    def report_result(self, proxy: Proxy, success: bool):
        """上报使用结果，更新统计"""
        if success:
            proxy.success_count += 1
        else:
            proxy.fail_count += 1
            if proxy.fail_count >= 10:
                self.blacklist.add(proxy.url)
                self.proxies.remove(proxy)

    def check_proxy(self, proxy: Proxy) -> bool:
        """检测代理是否可用"""
        try:
            resp = requests.get(
                self.check_url,
                proxies={"http": proxy.url, "https": proxy.url},
                timeout=5
            )
            proxy.last_check = time.time()
            return resp.status_code == 200
        except Exception:
            proxy.last_check = time.time()
            return False

    def health_check(self):
        """定期健康检查，剔除不可用IP"""
        for proxy in self.proxies[:]:
            if not self.check_proxy(proxy):
                proxy.fail_count += 3  # 检测失败加重惩罚
                if proxy.fail_count >= 10:
                    self.proxies.remove(proxy)

def fetch_with_retry(url: str, pool: ProxyPool, max_retries: int = 3) -> str | None:
    for attempt in range(max_retries):
        proxy = pool.get_proxy()
        if not proxy:
            print("没有可用代理了")
            return None
        
        try:
            resp = requests.get(
                url,
                proxies={"http": proxy.url, "https": proxy.url},
                timeout=10
            )
            if resp.status_code == 200:
                pool.report_result(proxy, success=True)
                return resp.text
            else:
                pool.report_result(proxy, success=False)
        except Exception as e:
            pool.report_result(proxy, success=False)
            print(f"尝试 {attempt+1} 失败: {e}")
    
    return None

if __name__ == "__main__":
    pool = ProxyPool()
    
    # 假设从API获取了一批代理IP
    raw_proxies = [
        ("1.2.3.4", 8080),
        ("5.6.7.8", 3128),
        # ...
    ]
    for ip, port in raw_proxies:
        pool.add_proxy(Proxy(ip=ip, port=port))
    
    # 先做一轮健康检查
    pool.health_check()
    print(f"可用代理数: {len(pool.proxies)}")
    
    # 使用
    result = fetch_with_retry("http://example.com", pool)
    if result:
        print("请求成功")

┌─────────────┐
│  代理源API  │  从供应商获取IP
└──────┬──────┘
       ▼
┌─────────────┐
│  清洗模块    │  去重、格式校验
└──────┬──────┘
       ▼
┌─────────────┐
│  检测模块    │  异步并发检测可用性
└──────┬──────┘
       ▼
┌─────────────┐
│  存储层      │  Redis（有序集合，按质量打分）
└──────┬──────┘
       ▼
┌─────────────┐
│  调度API     │  提供get/report接口
└──────┬──────┘
       ▼
┌─────────────┐
│  客户端SDK   │  自动重试、换IP、熔断
└─────────────┘

# 正确写法
proxies = {
    "http": "http://user:pass@1.2.3.4:8080",
    "https": "http://user:pass@1.2.3.4:8080"  # 注意：https代理也是写http://
}

session = requests.Session()
# 禁用连接复用
session.keep_alive = False