[网络协议]使用多线程技术批量查询随机生成的 IP 地址的归属地。

from threading import Thread,Lock
import requests
import json
from random import choice

class AddressAttribution(Thread):
    def __init__(self,id, ip):
        #super(AddressAttribution, self).__init__()
        Thread.__init__(self)

        self.ip = ip
        self.id = id

    def run(self):
        url = 'http://ip-api.com/json/%s' % self.ip
        page = requests.get(url).text
        dict_data = json.loads(page)
        lock.acquire()
        city = dict_data.get('city')
        country = dict_data.get('country')

        if (city and country):
            print(f"【id】:{self.id}  【ip】:{self.ip:>15}  【city】: {city:>20}  【country】: {country:>15}")

        lock.release()

if __name__ == '__main__':
    threads = []
    lock = Lock()

    for i in range(1, 1):
        r = range(1, 150)
        id = i
        ip = f'{choice(r)}.{choice(r)}.{choice(r)}.{choice(r)}'

        thread = AddressAttribution(id, ip)
        thread.start()

        threads.append(thread)

    [thread.join() for thread in threads]

    print("主线程...")

# 2、使用多线程技术批量 PING 同网段内所有设备。
from threading import Thread,Lock
from subprocess import Popen, PIPE, STDOUT

def ping(ip): 
    # -c compartment 路由隔离舱标识符。
    # -w timeout     等待每次回复的超时时间(毫秒)。
    response = Popen(["ping", "-c", "1", "-w", "10", ip.strip()],
    stdout=PIPE,
    stderr=STDOUT)
    stdout, stderr = response.communicate() 

    ls = str(stdout).split(r"\r\n")

    lines =str(stdout, 'gbk').split("\r\n")
    
    lock.acquire()

    if (response.returncode == 0 and len(lines) > 10):
        status = f"来自 {ip} 的回复: 可以访问目标主机。"
    else:
        status = f"来自 {ip} 的回复: 无法访问目标主机。"

    print(status) 

    lock.release()

if __name__ == '__main__':
    lock = Lock()
    net = '192.168.18.'
    print(f"打印网段的 {net}* 地址".center(50, '*'))
    threads = []

    for i in range(1, 10):
        ip = net + str(i)
        thread = Thread(target=ping, args=(ip,))
        thread.start()

    [thread.join() for thread in threads]