多线程ARP扫描函数中文件操作的优化问题求助

问题概述 #

实现多线程ARP扫描器时，在arp_scanner()线程函数内操作本地mac-vendors.csv文件匹配MAC厂商OUI，存在多线程重复读文件、资源竞争的问题；同时无法从线程函数中获取扫描结果，尝试全局变量也不是合理方案。

解决思路 #

• 预加载厂商数据：主线程启动前一次性读取mac-vendors.csv，将OUI与厂商名称映射存入字典，避免多线程重复打开文件
• 线程函数返回扫描结果：修改arp_scanner()，让它返回扫描到的设备IP、完整MAC及OUI，而非直接处理文件和打印
• 统一处理结果：主线程收集所有线程的扫描结果后，统一使用预加载的厂商字典匹配并输出，避免线程内的IO操作

修改后的代码 #

from scapy.all import srp, Ether, ARP
import socket
from concurrent.futures import ThreadPoolExecutor
import csv

def get_ip():
    dev_socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    dev_socket.connect(("8.8.8.8", 80)) 
    local_ip = str(dev_socket.getsockname()[0])
    base_ip = local_ip[:-3]
    return [f"{base_ip}{x}" for x in range(256)]

def load_vendor_data():
    # 主线程预加载厂商数据到字典
    vendor_map = {}
    with open('mac-vendors.csv', 'rt') as csv_file:
        csv_reader = csv.reader(csv_file, delimiter=',')
        for row in csv_reader:
            if len(row) >= 2:
                # 存储OUI的大写形式，避免匹配时重复转换
                vendor_map[row[0].upper()] = row[1]
    return vendor_map

def arp_scanner(target):
    arp_request = ARP(pdst=target)
    broadcast = Ether(dst="ff:ff:ff:ff:ff:ff")
    arp_request_broadcast = broadcast / arp_request
    answered_list = srp(arp_request_broadcast, timeout=1, verbose=False)[0]
    
    devices = []
    for element in answered_list:
        ip = element[1].psrc
        mac = element[1].hwsrc
        oui = mac[:8].upper()  # 直接提取OUI并转大写，无需正则处理字符串
        devices.append({
            "Local IP": ip,
            "MAC": mac,
            "OUI": oui
        })
    return devices

if __name__ == "__main__":
    # 预加载厂商数据
    vendor_map = load_vendor_data()
    ip_list = get_ip()
    
    # 使用线程池管理多线程，方便收集结果
    all_devices = []
    with ThreadPoolExecutor(max_workers=10) as executor:
        # 提交所有扫描任务
        futures = [executor.submit(arp_scanner, ip) for ip in ip_list]
        # 收集每个任务的结果
        for future in futures:
            result = future.result()
            if result:
                all_devices.extend(result)
    
    # 统一处理结果并输出
    for device in all_devices:
        print("----------------")
        print(f"Local IP: {device['Local IP']}")
        print(f"MAC: {device['MAC']}")
        # 匹配厂商
        vendor = vendor_map.get(device['OUI'], "Unknown Vendor")
        print(f"Vendor: {vendor}")
        print("----------------")

关键改进点 #

• 消除多线程文件竞争：load_vendor_data()在主线程启动前读取一次文件，生成OUI到厂商的映射字典，所有线程共享该内存数据，避免重复IO操作
• 简化数据提取逻辑：直接从Scapy返回的数据包中提取IP和MAC，无需将列表转字符串再用正则提取，减少不必要的字符串处理开销
• 高效收集线程结果：使用ThreadPoolExecutor替代手动创建线程，更方便安全地收集每个线程的扫描结果，完全摒弃全局变量方案
• 统一输出避免混乱：所有结果在主线程统一处理和打印，避免多线程同时打印导致的输出顺序错乱

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