Asyncio异步代理检测与同步版本结果差异问题求助

你的同步版代理检测脚本正常，但异步版几乎没有可用代理，核心问题出在代理URL格式错误和无限制并发触发限流两个点上，以下是具体分析和修复方案：

1. 代理URL格式不匹配 #

同步版requests的proxies参数允许省略协议头（仅写IP:Port），框架会自动补全适配；但aiohttp的proxy参数要求完整的代理URL。你筛选的是Https="yes"的代理，却用了http://开头的代理地址，导致代理无法正确转发HTTPS请求。

2. 无限制并发触发限流 #

同步版是串行请求，频率低，不容易触发目标网站或代理服务器的限流机制；异步版一次性发起所有代理测试请求，并发量过大，直接被目标网站拦截或代理服务器拒绝，导致大部分请求失败。

修改后的异步类代码 #

import asyncio
import aiohttp
from bs4 import BeautifulSoup
import time

class FreeProxyList(Proxy):
    def __init__(self, testEach=False, max_concurrent=10):
        self.testEach = testEach
        self.proxies = []
        self.session = None
        self.max_concurrent = max_concurrent  # 限制并发请求数

    async def setup(self):
        async with aiohttp.ClientSession() as session:
            async with session.get("https://free-proxy-list.net/") as response:
                html = await response.text()
                souped = BeautifulSoup(html, "html.parser")
                table = souped.find_all("table")[0]
                headers = [header.text for header in table.find_all("th")]

                proxies_list = []
                for row in table.find_all("tr")[1:]:
                    columns = row.find_all("td")
                    if columns:
                        proxy_info = {headers[i]: columns[i].text for i in range(len(headers))}
                        proxies_list.append(proxy_info)

                # 筛选HTTPS代理
                proxies_list = list(filter(lambda x: x['Https'] == "yes", proxies_list))
                # proxies_list = list(filter(lambda x: x['Code'] in ["US", "CA", "GB", "AU", "NZ"], proxies_list))

                if self.testEach:
                    self.proxies = []
                    # 使用Semaphore控制并发
                    semaphore = asyncio.Semaphore(self.max_concurrent)
                    tasks = [self.testOne(session, proxy, index, len(proxies_list), semaphore) for index, proxy in enumerate(proxies_list)]
                    results = await asyncio.gather(*tasks)
                    self.proxies = [proxy for proxy, result in zip(proxies_list, results) if result]

    async def testOne(self, session, proxy, index, total, semaphore):
        start = time.time()
        try:
            print(f"{index + 1} / {total :<5} Testing {proxy['IP Address']}:{proxy['Port']}")
            async with semaphore:
                # 修正为HTTPS协议的代理URL
                proxy_url = f"https://{proxy['IP Address']}:{proxy['Port']}"
                async with session.get(
                    "https://reedgraff.com", 
                    proxy=proxy_url, 
                    timeout=aiohttp.ClientTimeout(total=5)  # 使用aiohttp官方超时对象
                ) as response:
                    if response.status == 200:
                        return time.time() - start
                    else:
                        return None
        except KeyboardInterrupt:
            raise KeyboardInterrupt
        except:
            return None

关键修改点 #

• 新增max_concurrent参数，用asyncio.Semaphore限制并发请求数，避免触发目标网站或代理的限流
• 将代理URL修正为https://开头，匹配筛选出的HTTPS代理类型
• 替换timeout=5为aiohttp.ClientTimeout(total=5)，适配aiohttp的超时逻辑规范
• 优化导入语句位置，符合Python代码规范

内容的提问来源于stack exchange，提问作者Reed Graff