如何检查端口可用性？如何在端口检查器循环中用线程避免主线程阻塞？

嘿，我来帮你搞定这个异步端口扫描的问题！你的核心需求是让每个IP的扫描在单独线程里跑，不阻塞主线程对吧？咱们来一步步优化你的代码：

先把单个IP的扫描逻辑抽成独立函数，然后为每个IP启动一个线程，最后等待所有线程完成再统计结果：

import socket
import threading
import time

counting_open = []
counting_close = []
port = 25
hosts_file = "ip.txt"

# 单个IP的扫描逻辑，每个线程执行这个函数
def scan_single_host(host):
    # 跳过空行（避免ip.txt最后一行是空字符串导致报错）
    if not host.strip():
        return
    try:
        # 设置连接超时，防止单个IP扫描卡住太久
        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        sock.settimeout(1)  # 可根据网络环境调整超时时间
        # connect_ex返回0表示端口开放，非0表示关闭/无法连接
        result = sock.connect_ex((host, port))
        if result == 0:
            print(f"[+] {host}:{port} 开放")
            counting_open.append(host)
        else:
            print(f"[-] {host}:{port} 关闭")
            counting_close.append(host)
        sock.close()
    except Exception as e:
        print(f"[!] {host} 扫描出错: {str(e)}")

def scan():
    with open(hosts_file) as f:
        hosts = f.read().split("\n")
        threads = []
        for host in hosts:
            # 为每个IP创建线程，绑定扫描函数和参数
            thread = threading.Thread(target=scan_single_host, args=(host,))
            threads.append(thread)
            thread.start()
        
        # 等待所有扫描线程完成，主线程才继续执行后续统计
        for thread in threads:
            thread.join()
    
    # 扫描完成后输出统计结果
    print("\n=== 扫描结果统计 ===")
    print(f"开放端口的IP数量: {len(counting_open)}")
    print(f"关闭端口的IP数量: {len(counting_close)}")
    if counting_open:
        print("开放的IP列表:")
        for ip in counting_open:
            print(f"  {ip}")

if __name__ == "__main__":
    start_time = time.time()
    scan()
    print(f"\n扫描总耗时: {time.time() - start_time:.2f} 秒")

关键改动说明 #

• 拆分扫描逻辑：把单个IP的扫描抽成scan_single_host，让每个线程可以独立处理一个IP，避免主线程阻塞。
• 空行过滤：处理ip.txt中可能存在的空行，避免无效扫描。
• 超时设置：给socket添加超时时间，防止单个IP连接超时拖慢整体进度。
• 线程管理：用列表存储所有线程对象，最后通过join()等待所有线程完成，保证统计结果的完整性。

如果你的IP列表很长，手动创建大量线程会消耗过多系统资源，用ThreadPoolExecutor可以自动控制并发线程数，更高效：

import socket
import time
from concurrent.futures import ThreadPoolExecutor

counting_open = []
counting_close = []
port = 25
hosts_file = "ip.txt"
MAX_WORKERS = 10  # 控制并发线程数，可根据机器性能调整

def scan_single_host(host):
    if not host.strip():
        return
    try:
        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        sock.settimeout(1)
        result = sock.connect_ex((host, port))
        if result == 0:
            print(f"[+] {host}:{port} 开放")
            counting_open.append(host)
        else:
            print(f"[-] {host}:{port} 关闭")
            counting_close.append(host)
        sock.close()
    except Exception as e:
        print(f"[!] {host} 扫描出错: {str(e)}")

def scan():
    with open(hosts_file) as f:
        # 直接过滤空行，生成干净的IP列表
        hosts = [h.strip() for h in f.read().split("\n") if h.strip()]
    
    # 用线程池管理并发，自动处理线程的创建和销毁
    with ThreadPoolExecutor(max_workers=MAX_WORKERS) as executor:
        executor.map(scan_single_host, hosts)
    
    # 输出统计结果
    print("\n=== 扫描结果统计 ===")
    print(f"开放端口的IP数量: {len(counting_open)}")
    print(f"关闭端口的IP数量: {len(counting_close)}")
    if counting_open:
        print("开放的IP列表:")
        for ip in counting_open:
            print(f"  {ip}")

if __name__ == "__main__":
    start_time = time.time()
    scan()
    print(f"\n扫描总耗时: {time.time() - start_time:.2f} 秒")

额外注意事项 #

• 线程安全：列表的append操作在Python中是原子性的，所以多个线程同时修改counting_open和counting_close不会有问题；如果后续要做更复杂的列表操作（比如插入、删除），建议用threading.Lock加锁保护。
• 超时调整：如果是局域网扫描，超时可以设为0.5秒；公网扫描可调整为1-2秒，平衡速度和准确性。
• IP格式：确保ip.txt中每个IP单独占一行，代码已经做了strip()处理，即使有多余空格也会被自动去除。

内容的提问来源于stack exchange，提问作者Mostafa M. Mead