源码

# -*- encoding: utf-8 -*-
"""
@File    :   ping.py
@Time    :   2022-04-03 22:51
@Author  :   坐公交也用券
@Version :   1.0
@Contact :   faith01238@hotmail.com
@Homepage : https://liumou.site
@Desc    :   使用多线程进行ping操作
"""
import threading
import time
from subprocess import getstatusoutput
from os import path, getcwd, system

work = getcwd()
log = path.join(work, 'ping.txt')
txt = open(file=log, mode='w+', encoding='utf8')


def ping(ip):
    """
    开始Ping
    :param ip: 需要ping的IP
    :return:
    """
    c = f"ping {ip} -c 3"
    print(c)
    status = getstatusoutput(c)
    if status[0] == 0:
        result = f"{ip} : ok\n"
    else:
        result = f"{ip} : errors\n"
    print(result)
    txt.write(result)


class MyThread(threading.Thread):
    def __init__(self, ids, name, ip_list):
        threading.Thread.__init__(self)
        self.threadID = ids
        self.name = name
        self.ip = ip_list

    def run(self):
        print("开始线程：" + self.name)
        for ip_ in self.ip:
            print("开始ping: " + str(ip_))
            ping(ip=ip_)
        print("退出线程：" + self.name)


class PingCheck(object):
    def __init__(self, ip=[]):
        """
        开始进行Ping检查
        :param ip:
        """
        self.ip = ip

    def single(self):
        for ip_ in self.ip:
            ping(ip_)

    def threads(self):
        """
        使用多线程
        :return:
        """
        # 创建新线程
        thread1 = MyThread(1, "Thread-1", ip1)
        thread2 = MyThread(2, "Thread-2", ip2)
        thread3 = MyThread(3, "Thread-2", ip3)
        # 开启新线程
        thread1.start()
        thread2.start()
        thread3.start()

        thread1.join()
        thread2.join()
        thread3.join()
        print("退出主线程")


if __name__ == "__main__":
    start = time.time()
    ip = []
    ip1 = []
    ip2 = []
    ip3 = []
    for i in range(10, 19):
        ip1.append(f"10.1.1.{i}")

    for i in range(20, 29):
        ip2.append(f"10.1.1.{i}")

    for i in range(30, 39):
        ip3.append(f"10.1.1.{i}")

    for i in range(10, 39):
        ip.append(f"10.1.1.{i}")

    up = PingCheck(ip=ip)
    log = path.join(work, 'ping_single.txt')
    txt = open(file=log, mode='w+', encoding='utf8')
    start = time.time()
    up.single()
    end = time.time()
    sin_time = end - start

    start = time.time()
    log = path.join(work, 'ping_threads.txt')
    up.threads()
    end = time.time()
    t_time = end - start

    txt.close()

    print(f"单线程共耗时: {sin_time}")
    print(f"多线程共耗时: {t_time}")

效果

结论

使用线程数相比单线程方式得到的速度大概是1*N，其中N是启动的线程数量