Python批量发送大量TLS请求检测防火墙拦截域名的最优方式？

关于ThreadPoolExecutor是否是最快方案的结论 #

ThreadPoolExecutor不是该场景下的最优实现，200万级域名的TLS检测属于典型的高IO密集型任务，多线程方案受限于线程上下文切换开销，能承载的并发上限很低，效率远不如其他更适配的方案。

现有代码的可优化点 #

• 并发数设置太低：当前max_workers=20的配置跑200万域名耗时会极长，就算上调到几百，线程切换的额外开销也会快速抵消并发提升带来的收益。
• 文件IO操作过于频繁：每次检测到拦截域名就打开、关闭一次文件，会产生大量不必要的磁盘IO开销。
• 没有复用TCP连接：每次检测都新建到1.1.1.1的TCP连接，浪费了TCP握手的时间。

更高效的实现方案 #

1. 优先替换为异步IO方案 #

使用Python原生asyncio异步框架，单进程即可轻松承载上千并发，没有线程切换的额外开销，整体检测效率比ThreadPoolExecutor方案高3~10倍。参考实现代码如下：

import asyncio
import ssl
from collections import deque

# 配置项，可根据自身服务器性能调整
CONCURRENCY = 2000
SCAN_TIMEOUT = 1
TARGET_DNS_IP = "1.1.1.1"
OUTPUT_PATH = "blockedsites.txt"

ssl_context = ssl.create_default_context()
blocked_domains = deque()

async def scan_domain(domain: str, sem: asyncio.Semaphore):
    async with sem:
        try:
            # 先建立TCP连接
            reader, writer = await asyncio.wait_for(
                asyncio.open_connection(TARGET_DNS_IP, 443, ssl=None),
                timeout=SCAN_TIMEOUT
            )
            # 执行TLS握手
            await asyncio.wait_for(
                ssl_context.wrap_bio(reader, writer, server_hostname=domain),
                timeout=SCAN_TIMEOUT
            )
            writer.close()
            await writer.wait_closed()
        except asyncio.TimeoutError:
            return
        except ConnectionResetError:
            blocked_domains.append(domain)
            return
        except Exception:
            return

async def main(domains_list: list):
    semaphore = asyncio.Semaphore(CONCURRENCY)
    tasks = [scan_domain(d, semaphore) for d in domains_list]
    await asyncio.gather(*tasks)
    # 批量写入结果，仅产生一次磁盘IO
    with open(OUTPUT_PATH, "w", encoding="utf-8") as f:
        f.write("\n".join(blocked_domains))

if __name__ == "__main__":
    # 替换为你自己的200万域名列表
    sites = ['abc.com','def.com','geh.com']
    asyncio.run(main(sites))

2. 极致性能优化方案 #

如果需要更快的扫描速度，可以直接使用C语言开发的专业批量扫描工具，比如zgrab2，支持批量导入域名直接做TLS握手检测，性能比Python实现高一个量级，200万域名通常几十分钟就能扫描完成。

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