asyncio.run() 是唯一安全启动方式,须封装扫描逻辑为顶层 async 函数;应使用 asyncio.open_connection() 替代 socket.connect_ex(),配合 semaphore 限流、预解析 dns、三态端口判断(open/closed/filtered)以确保 python 3.12 下高效稳定运行。

asyncio.run() 必须包裹整个扫描入口
直接在模块顶层调用 asyncio.create_task() 或 asyncio.gather() 会触发 RuntimeError: no running event loop。Python 3.12 的事件循环策略更严格,不再容忍“裸奔”的协程调度。
正确做法是把所有扫描逻辑封装进一个顶层 async def scan_all_ports(...) 函数,然后用 asyncio.run(scan_all_ports(...)) 启动——这是唯一安全、可移植的启动方式。
- 别在
if __name__ == "__main__":下直接写await scan_port(...) - 别用
loop = asyncio.get_event_loop()(已弃用),Python 3.12 中该方法抛DeprecationWarning并将在 3.14 移除 - 如果要复用已有事件循环(比如嵌入 Flask/FastAPI),必须确认当前线程已运行循环,且用
asyncio.create_task()而非loop.create_task()
用 asyncio.open_connection() 替代 socket.connect_ex()
socket.connect_ex() 是同步阻塞调用,放进 asyncio.to_thread() 里跑只是“伪异步”:它仍占一个 OS 线程,无法突破线程数限制,还引入线程切换开销。Python 3.12 推荐路径是原生异步 I/O。
asyncio.open_connection() 是真正的异步连接尝试,底层复用 epoll/kqueue,不创建额外线程,超时由事件循环统一管理,响应更快、资源更省。
- 成功建立连接 → 端口开放,记得立刻
writer.close()避免连接堆积 - 抛
ConnectionRefusedError或OSError(含errno.ECONNREFUSED)→ 端口关闭 - 抛
asyncio.TimeoutError→ 端口可能被过滤(firewall drop),不是超时设置太短 - 别忽略
ssl=True参数:对 443/8443 这类端口,不设它会导致 TLS 握手失败后误判为关闭
并发量不是越多越好:控制 asyncio.Semaphore 限流
Python 3.12 默认事件循环(asyncio.DefaultEventLoopPolicy)在 macOS 和 Linux 上基于 epoll/kqueue,单机轻松支撑 10k+ 并发连接;但目标主机或中间网络设备(如防火墙、WAF)通常有连接速率限制。盲目开 5000 个协程,大概率触发 RST 洪水或 IP 封禁。
安全 Python 代码沙箱:在隔离的沙箱中执行任意 Python 代码,预装 requests、NumPy、pandas 库。返回 stdout、stderr。
用 asyncio.Semaphore(100) 是更稳妥的选择——这相当于每轮最多发起 100 个并发连接请求,既压满本地 I/O 能力,又避免激怒远端。
- 不要用
asyncio.gather(*tasks, return_exceptions=True)无限制吞任务,容易 OOM - 把
async with semaphore:放在每个scan_port()协程开头,不是只放在主循环里 - 配合
--timeout 1.0命令行参数传入,避免单个慢连接拖垮整批任务
端口状态判断不能只看异常类型
仅靠捕获 ConnectionRefusedError 判定“关闭”、靠成功连接判定“开放”,在真实网络中会漏判大量情况。例如:
- 某些 IDS/IPS 对高频 SYN 包直接静默丢弃,
asyncio.open_connection()会等满 timeout 后抛TimeoutError,此时应归为 “filtered”,而非 “closed” - 目标主机启用了 TCP Wrappers 或 xinetd,对未授权 IP 返回 RST,但实际服务存在——这种 RST 会被 Python 底层映射为
ConnectionRefusedError,和真关闭无异 - UDP 端口无法用此法扫描(
open_connection()仅支持 TCP),若需 UDP,必须切回socket+to_thread,且要自己构造 ICMP 错误解析逻辑
真正健壮的扫描器会把结果分三类:open、closed、filtered,并记录原始错误码(exc.errno)供后续分析。Python 3.12 的 Exception.__cause__ 和 traceback.print_exception() 支持更好,建议在 debug 模式下输出完整链路。
最易被忽略的是 DNS 解析阻塞——如果传入的是域名而非 IP,asyncio.open_connection(host, port) 内部会同步调用 getaddrinfo(),卡住整个协程。务必提前用 await loop.getaddrinfo(host, None) 或 asyncio.to_thread(socket.gethostbyname, host) 做预解析,否则并发再高也白搭。
Python免费学习笔记(深入):立即使用
在学习笔记中,你将探索 Python 的核心概念和高级技巧!










