企业级云爬虫架构中的代理IP池设计：高可用与自动扩容实现

import redis
import time

# 初始化Redis集群连接（生产哨兵模式）
rd = redis.RedisCluster(host="redis-proxy", port=6379, password="xxx")

def get_valid_proxy(biz: str, target_host: str):
    """
    落地级IP获取逻辑：防重复、防高频、优先优质IP
    """
    valid_key = f"proxy:pool:valid:{biz}"
    used_key = f"proxy:pool:used:{biz}"
    temp_ban_key = "proxy:pool:ban:temp"
    perm_ban_key = "proxy:pool:ban:perm"

    # 1. 随机取一个可用IP
    proxy_ip = rd.srandmember(valid_key)
    if not proxy_ip:
        return None, "资源池耗尽，触发扩容"

    proxy_ip = proxy_ip.decode()
    # 2. 过滤永久黑名单
    if rd.sismember(perm_ban_key, proxy_ip):
        rd.srem(valid_key, proxy_ip)
        return get_valid_proxy(biz, target_host)
    
    # 3. 禁止60s内同一IP访问同一站点（核心防封逻辑）
    used_info = rd.hget(used_key, proxy_ip)
    if used_info:
        last_host, last_time = used_info.decode().split(",")
        if last_host == target_host and time.time() - int(last_time) < 60:
            return get_valid_proxy(biz, target_host)

    # 4. 占用IP，写入使用记录
    rd.hset(used_key, proxy_ip, f"{target_host},{int(time.time())}")
    return proxy_ip, "success"

import requests
import time

def check_proxy_alive(ip: str, port: int) -> bool:
    """IP可用性探测，超时1.5s，适配爬虫低延迟要求"""
    proxies = {"http": f"http://{ip}:{port}", "https": f"http://{ip}:{port}"}
    try:
        res = requests.get("https://www.baidu.com", proxies=proxies, timeout=1.5)
        return res.status_code == 200
    except Exception:
        return False

def proxy_health_scan(biz: str):
    """定时巡检自愈逻辑，常驻后台线程"""
    valid_key = f"proxy:pool:valid:{biz}"
    temp_ban_key = "proxy:pool:ban:temp"
    perm_ban_key = "proxy:pool:ban:perm"
    fail_count = dict()

    while True:
        ip_list = rd.smembers(valid_key)
        for ip in ip_list:
            ip = ip.decode()
            if not check_proxy_alive(ip.split(":")[0], int(ip.split(":")[1])):
                fail_count[ip] = fail_count.get(ip, 0) + 1
                # 连续3次失败，永久剔除
                if fail_count[ip] >= 3:
                    rd.srem(valid_key, ip)
                    rd.sadd(perm_ban_key, ip)
                else:
                    # 临时失效，加入60s冷却
                    rd.zadd(temp_ban_key, {ip: time.time() + 60})
            else:
                fail_count[ip] = 0
        time.sleep(5)

def auto_scale_judge(biz: str) -> str:
    """
    自动扩缩容判定核心逻辑
    return: none/scale_up1/scale_up2/scale_up3/scale_down
    """
    valid_key = f"proxy:pool:valid:{biz}"
    used_key = f"proxy:pool:used:{biz}"

    total = rd.scard(valid_key)
    used = rd.hlen(used_key)
    if total == 0:
        return "scale_up3"
    
    usage = used / total
    ban_rate = get_recent_ban_rate(biz) # 近10s封禁率

    if usage >= 0.95 or ban_rate >= 0.08:
        return "scale_up3"
    elif usage >= 0.90:
        return "scale_up2"
    elif usage >= 0.85:
        return "scale_up1"
    elif usage <= 0.40:
        return "scale_down"
    return "none"