高并发数据采集中的代理池设计：从健康检查到智能调度 --- 辣椒HTTP实测攻略

import asyncio
import aiohttp
import time

class HealthChecker:
    def __init__(self, proxy_list, test_url="http://httpbin.org/ip", timeout=5):
        self.proxy_list = proxy_list
        self.test_url = test_url
        self.timeout = timeout
        self.healthy = {}

    async def _check_one(self, proxy):
        try:
            start = time.time()
            async with aiohttp.ClientSession() as session:
                async with session.get(
                    self.test_url,
                    proxy=f"http://{proxy}",
                    timeout=self.timeout
                ) as resp:
                    if resp.status == 200:
                        latency = time.time() - start
                        return proxy, True, latency
        except:
            pass
        return proxy, False, None

    async def check_all(self):
        tasks = [self._check_one(p) for p in self.proxy_list]
        results = await asyncio.gather(*tasks)
        for proxy, ok, latency in results:
            if ok:
                self.healthy[proxy] = latency
            else:
                self.healthy.pop(proxy, None)
        return self.healthy

class ProxyScorer:
    def __init__(self):
        self.success_count = {}
        self.fail_count = {}
        self.latency_avg = {}

    def record_success(self, proxy, latency):
        self.success_count[proxy] = self.success_count.get(proxy, 0) + 1
        old_avg = self.latency_avg.get(proxy, latency)
        self.latency_avg[proxy] = 0.8 * old_avg + 0.2 * latency  # 指数加权移动平均

    def record_fail(self, proxy):
        self.fail_count[proxy] = self.fail_count.get(proxy, 0) + 1

    def score(self, proxy):
        total = self.success_count.get(proxy, 0) + self.fail_count.get(proxy, 0)
        if total == 0:
            return 100
        success_rate = self.success_count.get(proxy, 0) / total
        latency = self.latency_avg.get(proxy, 1000)
        # 综合评分：成功率权重0.7，延迟权重0.3（经过归一化）
        score = success_rate * 70 + (1 - min(latency/1000, 1)) * 30
        return score

import random

def weighted_choice(proxy_list, scores):
    total = sum(scores[p] for p in proxy_list)
    rand = random.uniform(0, total)
    cum = 0
    for p in proxy_list:
        cum += scores[p]
        if rand <= cum:
            return p
    return proxy_list[0]

class ProxyScheduler:
    def __init__(self, proxy_list):
        self.health_checker = HealthChecker(proxy_list)
        self.scorer = ProxyScorer()

    async def get_proxy(self):
        await self.health_checker.check_all()
        healthy = list(self.health_checker.healthy.keys())
        if not healthy:
            return None
        scores = {p: self.scorer.score(p) for p in healthy}
        return weighted_choice(healthy, scores)

    async def fetch(self, url):
        proxy = await self.get_proxy()
        if not proxy:
            raise Exception("No available proxy")
        try:
            # 此处使用aiohttp或requests配合proxy发送请求
            # 为简化示例，省略具体请求代码
            # 假设成功，记录延迟
            self.scorer.record_success(proxy, 0.5)
            return response
        except Exception:
            self.scorer.record_fail(proxy)
            raise