使用Python测试代理IP池深度

为什么国内的动态住宅代理IP这么便宜，国外的这么贵，原来100个代理也可以称为动态住宅代理池。

国内的动态住宅代理市场因供应链完善、网络基础设施庞大且人口基数大，导致代理资源极其丰富，价格自然便宜。相反，国外代理尤其是欧美地区，由于用户分布分散、法律合规要求更严格、运营成本更高，价格相对昂贵。 代理IP的“动态住宅代理池”概念往往仅指拥有一定规模且可轮换的住宅IP集合。即使只有100个IP，如果这些IP能实现动态切换与高并发使用，也可视为一个“小型动态住宅代理池”。这种代理池在实际应用中可以通过异步测试与高频轮换，达到类似大规模代理池的效果。

一、高并发SOCKS5代理IP池测试工具介绍

本工具采用 Python 的 asyncio 异步库结合 aiohttp，实现高并发异步请求代理测试。通过模拟多个请求轮次，可快速检测代理池中实际可用的不同IP数量，评估代理池的健康状况与深度。

二、核心功能点及代码解析

1. 用户配置区（CONFIG）

这一部分集中配置测试所需的参数，方便统一管理和调整：

CONFIG = {
    "proxy_url": "socks5://user:pass@127.0.0.1:8080",  # SOCKS5代理地址，含账号密码
    "target_url": "http://httpbin.org/ip",              # 目标网址，用于获取当前出口IP
    "concurrency": 500,                                 # 单轮并发请求数量，越大并发越高
    "test_rounds": 10,                                  # 需要执行多少轮测试，模拟持续性能
    "timeout": 10,                                      # 每个请求最大超时时间（秒）
    "user_agents": [                                    # 多种User-Agent，模拟不同客户端请求
        "Mozilla/5.0 (Windows NT 10.0; Win64; x64)",
        "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7)",
        "Mozilla/5.0 (iPhone; CPU iPhone OS 15_0 like Mac OS X)"
    ]
}

此配置保证了测试的灵活性和真实感，例如随机User-Agent可以防止目标服务器简单封禁单一客户端请求特征。

2. 代理测试类 ProxyTester 设计

该类封装测试逻辑，管理状态和统计数据：

• results：保存所有请求的结果，包括成功和失败详情，便于后续分析。
• unique_ips：集合类型，自动去重，记录成功返回的不同代理IP。
• ip_counter：用字典统计每个IP被使用的次数，帮助了解IP分布及热门IP。
• success_count和fail_count：统计成功与失败请求总数，直接反映代理池健康度。
• session：aiohttp异步客户端会话，复用TCP连接，提升性能。

3. 异步会话初始化和关闭

async def init_session(self):
    # 限制TCP连接数，匹配并发配置，防止资源耗尽
    conn = aiohttp.TCPConnector(limit=CONFIG["concurrency"])
    self.session = aiohttp.ClientSession(
        connector=conn,
        timeout=aiohttp.ClientTimeout(total=CONFIG["timeout"])  # 设置超时时间，防止卡死
    )

async def close_session(self):
    if self.session:
        await self.session.close()  # 优雅关闭会话，释放资源

使用 TCPConnector 限制并发连接数，合理分配资源，防止因过多连接导致系统瓶颈或异常。

4. 核心请求函数 test_proxy

async def test_proxy(self, round_num: int, request_num: int):
    headers = {
        "User-Agent": random.choice(CONFIG["user_agents"])  # 每个请求随机User-Agent，伪装多客户端
    }

    proxy_url = CONFIG["proxy_url"]
    target_url = CONFIG["target_url"]

    try:
        async with self.session.get(
            target_url,
            proxy=proxy_url,
            headers=headers,
            ssl=False  # 关闭SSL验证，防止因证书问题请求失败
        ) as response:
            if response.status == 200:
                data = await response.json()
                ip = data.get("origin", "unknown")
                # httpbin.org返回可能包含多个IP，逗号分割，取第一个
                if "," in ip:
                    ip = ip.split(",")[0].strip()

                self.unique_ips.add(ip)  # 记录唯一IP
                self.ip_counter[ip] += 1
                self.success_count += 1

                return {
                    "round": round_num,
                    "request": request_num,
                    "ip": ip,
                    "status": "success"
                }
            else:
                self.fail_count += 1
                return {
                    "round": round_num,
                    "request": request_num,
                    "ip": None,
                    "status": f"http_error_{response.status}"
                }
    except Exception as e:
        self.fail_count += 1
        return {
            "round": round_num,
            "request": request_num,
            "ip": None,
            "status": str(e)
        }

通过捕获异常避免请求崩溃，ssl=False避免证书问题阻断请求。IP解析处理多IP返回，确保统计准确。

5. 批量执行一轮测试 run_test_round

async def run_test_round(self, round_num: int):
    tasks = []
    for i in range(1, CONFIG["concurrency"] + 1):
        task = asyncio.create_task(self.test_proxy(round_num, i))  # 创建异步任务
        tasks.append(task)

    round_results = await asyncio.gather(*tasks)  # 并发执行所有任务
    self.results.extend(round_results)

    # 轮次间微小延迟，防止请求瞬时爆发过猛
    await asyncio.sleep(0.1)

此处利用asyncio高并发优势，同时发起大量请求。延迟有助于避免目标服务器拒绝过载请求，模拟更真实访问。

6. 统计输出结果 print_stats

def print_stats(self):
    print("\n" + "="*60)
    print("代理池测试结果汇总".center(60))
    print("="*60)
    print(f"测试配置: {CONFIG['concurrency']}并发 × {CONFIG['test_rounds']}轮次")
    print(f"总请求次数: {self.success_count + self.fail_count}")
    print(f"成功请求: {self.success_count}")
    print(f"失败请求: {self.fail_count}")
    print(f"成功率: {self.success_count/(self.success_count+self.fail_count)*100:.2f}%")
    print(f"发现的唯一IP数量: {len(self.unique_ips)}")

    print("\nIP分布统计 (前20个):")
    sorted_ips = sorted(self.ip_counter.items(), key=lambda x: x[1], reverse=True)
    for ip, count in sorted_ips[:20]:
        print(f"{ip:<18} : {count}次")

    if len(self.ip_counter) > 20:
        print(f"...(共 {len(self.ip_counter)} 个不同IP)")

输出详细统计数据，方便快速判断代理池有效IP数量及热门IP情况，便于后续优化代理选择。

5. 完整的代码

import asyncio
import aiohttp
from collections import defaultdict
import time
from typing import List, Dict
import random

# ======== 用户配置区域 ========
CONFIG = {
    "proxy_url": "socks5://user:pass@127.0.0.1:8080",  # SOCKS5代理地址
    "target_url": "http://httpbin.org/ip",                      # 测试目标URL
    "concurrency": 500,                                        # 并发请求数量
    "test_rounds": 10,                                         # 测试轮次
    "timeout": 10,                                             # 单个请求超时(秒)
    "user_agents": [                                           # 随机User-Agent
        "Mozilla/5.0 (Windows NT 10.0; Win64; x64)",
        "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7)",
        "Mozilla/5.0 (iPhone; CPU iPhone OS 15_0 like Mac OS X)"
    ]
}
# =============================

class ProxyTester:
    def __init__(self):
        self.results = []
        self.unique_ips = set()
        self.ip_counter = defaultdict(int)
        self.success_count = 0
        self.fail_count = 0
        self.session = None

    async def init_session(self):
        conn = aiohttp.TCPConnector(limit=CONFIG["concurrency"])
        self.session = aiohttp.ClientSession(
            connector=conn,
            timeout=aiohttp.ClientTimeout(total=CONFIG["timeout"])
        )

    async def close_session(self):
        if self.session:
            await self.session.close()

    async def test_proxy(self, round_num: int, request_num: int):
        headers = {
            "User-Agent": random.choice(CONFIG["user_agents"])
        }
        
        proxy_url = CONFIG["proxy_url"]
        target_url = CONFIG["target_url"]
        
        try:
            async with self.session.get(
                target_url,
                proxy=proxy_url,
                headers=headers,
                ssl=False
            ) as response:
                if response.status == 200:
                    data = await response.json()
                    ip = data.get("origin", "unknown")
                    if "," in ip:  # 处理可能有多个IP的情况
                        ip = ip.split(",")[0].strip()
                    
                    self.unique_ips.add(ip)
                    self.ip_counter[ip] += 1
                    self.success_count += 1
                    
                    return {
                        "round": round_num,
                        "request": request_num,
                        "ip": ip,
                        "status": "success"
                    }
                else:
                    self.fail_count += 1
                    return {
                        "round": round_num,
                        "request": request_num,
                        "ip": None,
                        "status": f"http_error_{response.status}"
                    }
        except Exception as e:
            self.fail_count += 1
            return {
                "round": round_num,
                "request": request_num,
                "ip": None,
                "status": str(e)
            }

    async def run_test_round(self, round_num: int):
        tasks = []
        for i in range(1, CONFIG["concurrency"] + 1):
            task = asyncio.create_task(self.test_proxy(round_num, i))
            tasks.append(task)
        
        round_results = await asyncio.gather(*tasks)
        self.results.extend(round_results)
        
        # 每个请求之间添加微小延迟，避免瞬时爆发
        await asyncio.sleep(0.1)

    def print_stats(self):
        print("\n" + "="*60)
        print("代理池测试结果汇总".center(60))
        print("="*60)
        print(f"测试配置: {CONFIG['concurrency']}并发 × {CONFIG['test_rounds']}轮次")
        print(f"总请求次数: {self.success_count + self.fail_count}")
        print(f"成功请求: {self.success_count}")
        print(f"失败请求: {self.fail_count}")
        print(f"成功率: {self.success_count/(self.success_count+self.fail_count)*100:.2f}%")
        print(f"发现的唯一IP数量: {len(self.unique_ips)}")
        
        print("\nIP分布统计 (前20个):")
        sorted_ips = sorted(self.ip_counter.items(), key=lambda x: x[1], reverse=True)
        for ip, count in sorted_ips[:20]:
            print(f"{ip:<18} : {count}次")
        
        if len(self.ip_counter) > 20:
            print(f"...(共 {len(self.ip_counter)} 个不同IP)")

async def main():
    tester = ProxyTester()
    await tester.init_session()
    
    start_time = time.time()
    
    print(f"\n开始高并发代理测试: {CONFIG['concurrency']}并发 × {CONFIG['test_rounds']}轮次")
    print(f"目标URL: {CONFIG['target_url']}")
    print(f"代理地址: {CONFIG['proxy_url']}\n")
    
    for round_num in range(1, CONFIG["test_rounds"] + 1):
        print(f"正在执行第 {round_num}/{CONFIG['test_rounds']} 轮测试...")
        await tester.run_test_round(round_num)
    
    await tester.close_session()
    
    tester.print_stats()
    
    total_time = time.time() - start_time
    print(f"\n总测试时间: {total_time:.2f}秒")
    print(f"平均每秒请求: {(tester.success_count + tester.fail_count)/total_time:.2f}")
    print("测试完成\n")

if __name__ == "__main__":
    # 检查依赖
    try:
        import aiohttp
        import aiohttp_socks
    except ImportError:
        print("\n错误: 请先安装依赖库")
        print("安装命令: pip install aiohttp aiohttp_socks\n")
        exit(1)
    
    # 设置事件循环策略(Windows需要)
    if hasattr(asyncio, "WindowsSelectorEventLoopPolicy"):
        asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())
    
    asyncio.run(main())

细节说明：输出详细统计数据，方便快速判断代理池有效IP数量及热门IP情况，便于后续优化代理选择。

三、运行示例及详细输出

运行程序时，终端会打印每轮测试进度，最终输出结果示例如下：

开始高并发代理测试: 500并发 × 10轮次
目标URL: http://httpbin.org/ip
代理地址: socks5://user:pass@127.0.0.1:8080

正在执行第 1/10 轮测试...
正在执行第 2/10 轮测试...
...
正在执行第 10/10 轮测试...

============================================================
                      代理池测试结果汇总                      
============================================================
测试配置: 500并发 × 10轮次
总请求次数: 5000
成功请求: 4600
失败请求: 400
成功率: 92.00%
发现的唯一IP数量: 120

IP分布统计 (前20个):
192.168.0.101     : 350次
192.168.0.102     : 320次
192.168.0.103     : 300次
...
...(共 120 个不同IP)

总测试时间: 25.37秒
平均每秒请求: 197.06
测试完成
    

四、总结

本测试工具利用异步编程实现高并发请求，显著提升代理池检测效率。通过统计唯一IP数量和请求成功率，能直观反映代理池的真实可用性和动态分布。

理解“动态住宅代理池”不仅在于IP数量，更关键是IP的真实性和动态更换能力。一个仅含100个不同IP的代理池，如果能频繁切换且均为住宅宽带IP，也可被称为动态住宅代理池。

结合国内外价格差异，国内代理供应量大，运营成本低，导致价格便宜；而国外代理因资源稀缺、维护复杂，价格自然较高。

因此，评估代理池质量需要结合多维度数据，工具的高并发检测为代理池运营和使用提供了有效手段。