使用python多线程检测ip地址

#检测得ip地址在C:\Users\admin\Desktop\iplist.txt文件中

import subprocess
import threading
import IPy
import re
from queue import Queue
import time

ip_use = []   #检测使用IP地址
ip_unused = []  #检测未使用ip地址

WORK_THREAD = 100   #定义100个工作线程
IP_QUEUE = Queue()  #定义队列

#获取到得ip地址使用排除网络号和主机号
def get_ip_by_mask(mask):
    ip_list = []
    for j in IPy.IP(i):
        ip = str(j)
        result = re.search(r'(\.0)$|(\.255)$', ip)
        if result == None:
            ip_list.append(ip)
    return ip_list

#ping检测ip地址
def ip_ping():
    global ip_use
    global ip_unused
    while not IP_QUEUE.empty():
        ip = IP_QUEUE.get()
        out = subprocess.run('ping -n 2  -w 5 %s' % ip, shell=True, stdout=subprocess.PIPE)
        if out.returncode == 0:
            print('{}地址可达'.format(ip))
            ip_use.append(ip)

        else:
            print('{}地址不可达'.format(ip))
            ip_unused.append(ip)



if __name__ == '__main__':
    f = open(r'C:\Users\admin\Desktop\iplist.txt', 'r')
    ip_list = []
    for i in f.readlines():
        mask = i.strip().strip('\r').strip('\n')
        ip_list += get_ip_by_mask(mask)
    for i in ip_list:
        IP_QUEUE.put(i)

    threads = []
    start_time = time.time()
    for i in range(WORK_THREAD):
        thread = threading.Thread(target=ip_ping)
        thread.start()
        threads.append(thread)
    for thread in threads:
        thread.join()
    print('程序运行耗时：%s' % (time.time() - start_time))
    print(ip_use)
    print(ip_unused)