Python 3 Multiprocessing Pool端口扫描问题求助
Python Multiprocessing Pool 端口扫描问题修复与优化
我帮你梳理下代码里的问题,然后给出修复后的完整实现,同时做了一些实用优化:
原代码里的核心问题
- 语法错误:
s.connect((target,port)这里少了一个右括号,会直接导致代码运行报错 - 全局列表
list1的使用没必要,而且在多进程场景下全局变量容易引发意外问题,不如直接生成端口列表 - 异常处理过于宽泛,直接
except:会捕获所有异常,不利于排查具体问题,应该针对socket相关异常做针对性处理 - 多进程Pool的代码没写完,
with Po...需要补全为with Pool() as p:来正确管理进程池生命周期 - 多进程下直接打印会导致输出混乱,多个进程同时打印会让日志变得杂乱无章
修复并优化后的完整代码
from multiprocessing import Pool import socket target = 'hackthissite.org' def scan_port(port): s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) # 设置超时时间,避免单个端口连接卡住拖慢整体扫描 s.settimeout(0.5) try: s.connect((target, port)) s.close() return port # 返回成功端口,而非直接打印 except (socket.timeout, ConnectionRefusedError): # 捕获最常见的连接超时、连接拒绝异常,不做冗余输出 s.close() return None except Exception as e: s.close() print(f"Unexpected error on port {port}: {str(e)}") return None def get_port_range(): try: port_range = int(input("How many ports would you like to scan? ")) return list(range(1, port_range + 1)) except ValueError: print("Please enter a valid integer!") return [] if __name__ == '__main__': ports = get_port_range() if not ports: exit() # 进程池默认使用CPU核心数,也可以手动指定比如Pool(15)调整并发数 with Pool() as p: # imap_unordered能在进程完成后立即获取结果,比map等待全部完成更高效 results = p.imap_unordered(scan_port, ports) # 过滤并收集开放的端口 open_ports = [port for port in results if port is not None] # 统一打印结果,避免多进程输出混乱 if open_ports: print("\nOpen ports found:") for port in sorted(open_ports): print(f"✅ Port {port} is open") else: print("\nNo open ports found in the scanned range.")
优化细节说明
- 把模糊的函数名
f改成scan_port,让代码可读性更强 - 给socket设置超时时间,避免某个端口的连接尝试卡住整个扫描流程
- 拆分异常处理逻辑,只对意外异常打印日志,减少无用输出
- 移除全局列表,改用函数返回端口列表,让代码更模块化、更安全
- 使用
imap_unordered替代map,能更快获取已完成的扫描结果 - 最后统一整理并排序打印开放端口,解决多进程同时打印导致的输出混乱问题
内容的提问来源于stack exchange,提问作者Walker Ward
相关产品推荐
相关产品推荐

