请求协助：为Python Ping代码实现多进程/多线程并控制并发数

Hey there! 我帮你把这段Ping检测代码改成支持自定义并发数的版本，分别提供多线程和多进程两种实现方案——考虑到Ping属于IO密集型操作，多线程通常会更高效，不过两种方案我都写出来了，你可以按需选择。

多线程实现版本（推荐） #

这个版本用ThreadPoolExecutor来控制并发数，非常适合Ping这类IO密集型任务，开销小、效率高：

import subprocess
import csv
from concurrent.futures import ThreadPoolExecutor, as_completed

def ping(hostname):
    """执行单次Ping检测，返回主机名和检测状态"""
    try:
        # Windows环境下的Ping命令参数，Linux/macOS请改成 ["ping", "-c", "1", "-W", "1", hostname]
        p = subprocess.Popen(
            ["ping", "-n", "1", "-w", "1000", hostname],
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            creationflags=subprocess.CREATE_NO_WINDOW  # 隐藏Windows命令行窗口（可选）
        )
        stdout, _ = p.communicate()
        output = stdout.rstrip().decode('UTF-8')
        # 判断Ping结果：不可达或返回码非0都标记为失败
        if "unreachable." in output or p.returncode != 0:
            return (hostname, "failed")
        return (hostname, "ok")
    except Exception as e:
        return (hostname, f"error: {str(e)}")

def batch_ping(hosts, max_concurrent=50):
    """批量执行Ping检测，通过max_concurrent参数控制并发数"""
    results = []
    # 创建线程池，max_workers就是你要设置的并发数
    with ThreadPoolExecutor(max_workers=max_concurrent) as executor:
        # 提交所有Ping任务到线程池
        future_to_host = {executor.submit(ping, host): host for host in hosts}
        # 遍历完成的任务，实时收集结果
        for future in as_completed(future_to_host):
            host = future_to_host[future]
            try:
                result = future.result()
                results.append(result)
                print(f"Ping {host}: {result[1]}")
            except Exception as e:
                print(f"Ping {host} 出现异常: {str(e)}")
                results.append((host, f"exception: {str(e)}"))
    return results

def save_results_to_csv(results, filename="ping_results.csv"):
    """将检测结果保存到CSV文件，方便后续分析"""
    with open(filename, mode='w', newline='', encoding='utf-8') as f:
        writer = csv.writer(f)
        writer.writerow(["Hostname", "Status"])
        writer.writerows(results)

if __name__ == "__main__":
    # 示例主机列表，你可以替换成自己的列表（比如从txt文件读取）
    test_hosts = [f"192.168.1.{i}" for i in range(1, 101)]
    # 设置并发数为50，你可以根据需求调整
    ping_results = batch_ping(test_hosts, max_concurrent=50)
    # 保存结果到CSV
    save_results_to_csv(ping_results)
    print("Ping检测完成，结果已保存到ping_results.csv")

关键点说明： #

• 直接通过max_concurrent参数设置并发数，比如设为50就会同时运行50个Ping任务
• 保留了原有的Ping逻辑，同时增加了异常处理，避免单个任务失败影响整体批量检测
• 针对Windows环境添加了隐藏命令行窗口的参数，不需要可以直接删掉
• 支持将结果保存到CSV，方便后续统计或分析

多进程实现版本（仅作参考） #

如果你的Ping逻辑后续会加入CPU密集型处理，可以用这个版本，不过对于纯Ping操作，多进程的开销会比多线程大：

import subprocess
import csv
from concurrent.futures import ProcessPoolExecutor, as_completed

def ping(hostname):
    """执行单次Ping检测，返回主机名和检测状态"""
    try:
        # Windows环境下的Ping命令参数，Linux/macOS请改成 ["ping", "-c", "1", "-W", "1", hostname]
        p = subprocess.Popen(
            ["ping", "-n", "1", "-w", "1000", hostname],
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            creationflags=subprocess.CREATE_NO_WINDOW  # 隐藏Windows命令行窗口（可选）
        )
        stdout, _ = p.communicate()
        output = stdout.rstrip().decode('UTF-8')
        # 判断Ping结果：不可达或返回码非0都标记为失败
        if "unreachable." in output or p.returncode != 0:
            return (hostname, "failed")
        return (hostname, "ok")
    except Exception as e:
        return (hostname, f"error: {str(e)}")

def batch_ping(hosts, max_concurrent=50):
    """批量执行Ping检测，通过max_concurrent参数控制并发数"""
    results = []
    # 创建进程池，max_workers就是你要设置的并发数
    with ProcessPoolExecutor(max_workers=max_concurrent) as executor:
        # 提交所有Ping任务到进程池
        future_to_host = {executor.submit(ping, host): host for host in hosts}
        # 遍历完成的任务，实时收集结果
        for future in as_completed(future_to_host):
            host = future_to_host[future]
            try:
                result = future.result()
                results.append(result)
                print(f"Ping {host}: {result[1]}")
            except Exception as e:
                print(f"Ping {host} 出现异常: {str(e)}")
                results.append((host, f"exception: {str(e)}"))
    return results

def save_results_to_csv(results, filename="ping_results.csv"):
    """将检测结果保存到CSV文件，方便后续分析"""
    with open(filename, mode='w', newline='', encoding='utf-8') as f:
        writer = csv.writer(f)
        writer.writerow(["Hostname", "Status"])
        writer.writerows(results)

if __name__ == "__main__":
    # 示例主机列表，你可以替换成自己的列表（比如从txt文件读取）
    test_hosts = [f"192.168.1.{i}" for i in range(1, 101)]
    # 设置并发数为50，你可以根据需求调整
    ping_results = batch_ping(test_hosts, max_concurrent=50)
    # 保存结果到CSV
    save_results_to_csv(ping_results)
    print("Ping检测完成，结果已保存到ping_results.csv")

注意事项： #

• 多进程版本和多线程版本的核心逻辑几乎一致，只是把ThreadPoolExecutor换成了ProcessPoolExecutor
• Linux/macOS环境下需要修改Ping命令的参数，把-n换成-c，-w换成-W
• 并发数不要设置得过高，否则可能会触发系统的网络限制或导致网络拥堵，50是比较合理的数值，可根据实际情况调整

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