基于asyncio的批量IP Ping监控性能不佳问题及优化建议

核心问题 #

并发100个Ping任务检测1024个IP时，部分IP无响应（串行检测正常，Wireshark显示Ping包已发出但无回复），添加0.08秒任务创建延迟后问题缓解，但需找到根本原因并优化性能。

可能的根本原因 #

• 系统ICMP速率限制：多数操作系统（Linux/Windows）默认对ICMP Echo请求的发送频率有配额限制，并发大量发送会触发限流，导致部分包被丢弃或延迟。
• 子进程资源过载：通过create_subprocess_shell调用系统ping命令，每个任务都要创建独立子进程，100个并发子进程会占用大量系统资源（进程表、内存、CPU），导致部分进程无法及时完成ICMP交互。
• 网络设备限流：目标网络的路由器、防火墙或目标设备本身可能配置了ICMP速率限制，并发Ping请求被设备主动丢弃。
• 任务创建过于集中：一次性创建1024个任务，即使有信号量控制100个并发，短时间内集中启动的100个Ping进程会瞬间发送大量ICMP包，超出网络或系统的处理能力。

优化方案 #

1. 替换系统Ping为原生ICMP库（最有效） #

放弃调用系统ping子进程，改用Python原生ICMP异步库（如icmplib）直接发送ICMP包，彻底消除子进程的资源开销。

示例代码：

from icmplib import async_ping
from functools import wraps
import asyncio

def wait_resources(concurrent_tasks=1):
    semaphore = asyncio.Semaphore(concurrent_tasks)
    def wrapper(function):
        @wraps(function)
        async def inside_function(*args, **kwargs):
            async with semaphore:
                return await function(*args, **kwargs)
        return inside_function
    return wrapper

@wait_resources(concurrent_tasks=100)
async def asyncping(ip):
    try:
        # 发送1次Ping，超时1秒
        host = await async_ping(ip, count=1, timeout=1, privileged=False)
        return host.is_alive
    except:
        return False

async def run():
    ips_to_validate = get_all_ips()
    up_devices = 0
    # 用asyncio.as_completed迭代完成的任务，避免一次性创建所有任务
    tasks = [asyncio.create_task(asyncping(ip), name=ip) for ip in ips_to_validate]
    for task in asyncio.as_completed(tasks):
        result = await task
        if result:
            up_devices += 1
    print(f'devices: {up_devices}/{len(ips_to_validate)}')

2. 精细化速率控制 #

不要用固定延迟，改用令牌桶算法控制ICMP包的发送速率，避免短时间内集中发包。

示例：添加全局速率控制锁，每秒最多发送N个请求：

import time

rate_limit = 50  # 每秒最多50个Ping请求
last_time = time.time()
token_count = rate_limit
rate_lock = asyncio.Lock()

async def acquire_token():
    global token_count, last_time
    async with rate_lock:
        now = time.time()
        # 补充令牌
        token_count += int((now - last_time) * rate_limit)
        if token_count > rate_limit:
            token_count = rate_limit
        last_time = now
        # 等待令牌
        while token_count <= 0:
            await asyncio.sleep(0.01)
            now = time.time()
            token_count += int((now - last_time) * rate_limit)
            last_time = now
        token_count -= 1

# 修改asyncping函数，先获取令牌再执行
@wait_resources(concurrent_tasks=100)
async def asyncping(ip):
    await acquire_token()
    # 后续Ping逻辑...

3. 优化系统ICMP参数（Linux） #

临时调整系统ICMP速率限制（谨慎使用，避免影响其他应用）：

# 关闭ICMP速率限制
sysctl -w net.ipv4.icmp_ratelimit=0
# 调整ICMP掩码（允许所有ICMP类型不受限）
sysctl -w net.ipv4.icmp_ratemask=0

4. 改进任务管理 #

避免一次性创建所有1024个任务，改为分批处理，减少内存占用和瞬时压力：

async def run():
    ips_to_validate = list(get_all_ips())
    batch_size = 100
    up_devices = 0
    total = len(ips_to_validate)
    # 分批处理IP
    for i in range(0, total, batch_size):
        batch = ips_to_validate[i:i+batch_size]
        tasks = [asyncio.create_task(asyncping(ip), name=ip) for ip in batch]
        for task in asyncio.as_completed(tasks):
            result = await task
            if result:
                up_devices += 1
    print(f'devices: {up_devices}/{total}')

5. 优化Ping参数 #

精简Ping命令的参数，缩短单个任务的执行时间：

# 原命令优化：减少Ping次数为1，超时1秒，去掉间隔参数
cmd = f'ping -c 1 -W 1 {ip}'

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