Python结合异步与多线程实现高效主机扫描方案问询

需要开发一款支持自定义超时的主机扫描程序，具备ICMP Ping、端口扫描及HTTP信息获取功能，要最大化利用CPU提升效率。仅主线程异步无法满足高负载需求，目标是60秒内完成10万台主机扫描，且超时标记为DOWN的主机占比控制在0.05%（0.0005）。

想知道如何创建多线程并在其中执行批量主机的异步扫描，也欢迎其他解决方案。

用户现有代码 #

import asyncio
import os
import socket
import aioping
import concurrent.futures

async def ping_host(host, timeout=5, retry=1):
    for i in range(retry):
        try:
            delay = await aioping.ping(host, timeout=timeout)
            return True, host, delay
        except (OSError, asyncio.TimeoutError, socket.gaierror):
            pass
    return False, host


def ping_hosts(hosts):
    results = {}
    with concurrent.futures.ThreadPoolExecutor(max_workers=os.cpu_count()) as pool:
        loop = asyncio.get_event_loop()
        tasks = []
        for host in hosts:
            task = loop.run_in_executor(pool, ping_host, host)
            tasks.append(task)
        for index, task in enumerate(asyncio.as_completed(tasks)):
            host = hosts[index]
            try:
                result = loop.run_until_complete(task)
            except Exception as e:
                print(f"Ex: {e}")
                result = False
            results[host] = result
    return results

报错信息 #

Ex: This event loop is already running
...
google.com: DOWN
False (RESULT)
...

问题分析与解决方案 #

现有代码核心问题 #

当前代码错误在于在同一个事件循环里混用run_in_executor和run_until_complete，且run_in_executor直接将异步函数丢到线程池执行——异步函数必须在事件循环中运行，线程池无默认事件循环，直接调用仅返回协程对象，不会执行，最终引发事件循环冲突和结果异常。

正确的多线程+异步实现 #

每个线程独立维护一个事件循环，负责批量执行异步扫描任务，既用多线程占满CPU，又用异步处理IO密集型扫描操作，最大化效率。

优化后代码

import asyncio
import os
import socket
import aioping
from concurrent.futures import ThreadPoolExecutor

async def ping_host(host, timeout=5, retry=1):
    for i in range(retry):
        try:
            delay = await aioping.ping(host, timeout=timeout)
            return True, host, delay
        except (OSError, asyncio.TimeoutError, socket.gaierror):
            pass
    return False, host

# 单个线程内的批量异步扫描逻辑
async def batch_ping(hosts, timeout=5, retry=1):
    tasks = [ping_host(host, timeout, retry) for host in hosts]
    return await asyncio.gather(*tasks)

# 线程入口：创建独立事件循环并执行批量任务
def thread_worker(host_batch, timeout, retry):
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)
    results = loop.run_until_complete(batch_ping(host_batch, timeout, retry))
    loop.close()
    return results

def ping_hosts(hosts, timeout=5, retry=1, max_workers=None):
    if max_workers is None:
        max_workers = os.cpu_count()
    
    # 将主机列表拆分到多个批次，每个线程处理一个批次
    batch_size = (len(hosts) + max_workers - 1) // max_workers
    host_batches = [hosts[i:i+batch_size] for i in range(0, len(hosts), batch_size)]
    
    results = {}
    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        # 提交所有批次任务到线程池
        futures = [executor.submit(thread_worker, batch, timeout, retry) for batch in host_batches]
        
        # 汇总所有线程的扫描结果
        for future in futures:
            batch_results = future.result()
            for res in batch_results:
                if isinstance(res, tuple):
                    status, host, *delay = res
                    results[host] = (status, delay[0] if delay else None)
                else:
                    results[res[1]] = res
    
    return results

关键优化点 #

• 线程独立事件循环：避免事件循环冲突，每个线程自主调度异步任务
• 批量任务拆分：将10万台主机平均分配到各线程，平衡负载
• 异步批量执行：用asyncio.gather批量处理Ping任务，减少调度开销
• 参数可配置：支持自定义超时、重试次数、线程数，适配不同场景

其他效率优化建议 #

1. 超时与重试策略调整：

   • 针对10万台主机的目标，建议将Ping超时设为1-2秒，重试1次即可，避免长时间等待拖慢整体进度
   • 可根据网络环境动态调整超时，比如对响应快的网段用更短超时

2. 端口扫描与HTTP获取的异步实现：

   • 端口扫描用asyncio.open_connection实现异步TCP连接
   • HTTP信息获取用aiohttp库发送异步请求，可与Ping任务整合到同一异步批量函数中

3. 结果处理优化：

   • 用队列异步收集结果，避免主线程等待所有任务完成才处理
   • 对标记为DOWN的主机做二次验证，确保误判率控制在0.05%以内

4. 系统资源调优：

   • 调整系统文件描述符上限（Linux下用ulimit -n），避免因打开过多Socket报错
   • 线程数不要超过CPU核心数的2倍，避免上下文切换开销

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