Python线程内存泄漏排查：LAN设备Ping检测程序内存飙升问题

问题背景 #

使用threading实现的LAN设备Ping检测程序，预期内存占用低于80MB，但运行5小时后内存占用超过3GB。查阅Python Bug追踪器发现这是用户常见反馈，但现有解决方案可信度不足。尝试获取线程内存地址进行删除操作未果，推测Python不允许直接访问线程内存地址。

核心代码 #

def update_ping(self, data, ping_queue):
        result_queue = queue.Queue()
        local = threading.local()
        while not self.stop_threads:
            threads = []
            for seccion in data:
                lineas = seccion["lineas"]
                for linea in lineas:
                    equipos = linea["equipos"]
                    for equipo in equipos:
                        ip = equipo["ip"]
                        thread = threading.Thread(target=self.analyze_device, args=(ip, result_queue), daemon=False)
                        threads.append(thread)
                        thread.start()
            for thread in threads:
                thread.join()
            new_data = []
            print(f"Tamaño cola:{result_queue.qsize()}")
            while not result_queue.empty():
                ip, status_color = result_queue.get()
                for seccion in data:
                    lineas = seccion["lineas"]
                    for linea in lineas:
                        equipos = linea["equipos"]
                        for equipo in equipos:
                            if equipo["ip"] == ip:
                                equipo["status_color"] = status_color
            for seccion in data:
                nuevos_equipos_seccion = []
                lineas = seccion["lineas"]
                for linea in lineas:
                    nuevos_equipos_linea = []
                    equipos = linea["equipos"]
                    for equipo in equipos:
                        ip = equipo["ip"]
                        status_color = equipo["status_color"]
                        nuevos_equipos_linea.append(equipo)
                    linea["equipos"] = nuevos_equipos_linea
                    nuevos_equipos_seccion.append(linea)
                seccion["lineas"] = nuevos_equipos_seccion
                new_data.append(seccion)
            ping_queue.put(new_data)
            print("Ping realizado")
            print(f"Recolección: {gc.collect()}")
            print(f"Depuradores: {gc.get_debug()}")
            print(f"Basura: {gc.garbage}")


    def analyze_device(self, ip, result_queue):
        try:
            if self.check_online(ip):
                status_color = "green"
            else:
                status_color = "red"
            
            result_queue.put((ip, status_color))
            return
        except Exception as e:
            print(f"Error en el hilo {ip}: {str(e)}")
    
    def check_online(self, ip):
        check = False
        for _ in range(5):
            data = ping(ip)
            if isinstance(data, float):
                check = True
                break
            else:
                time.sleep(0.5)
        return check

内存泄漏修复方案 #

• 优化线程生命周期管理：放弃每次循环创建新线程的方式，改用concurrent.futures.ThreadPoolExecutor线程池复用线程，避免频繁创建销毁线程带来的内存碎片与对象残留；同时将线程设为daemon=True或由线程池自动管理，确保线程能被正确回收。
• 清理冗余数据操作：
  • threading.local()只需在循环外初始化一次，无需每次循环重复创建；
  • 直接修改原data中的status_color字段，无需构建多层嵌套的new_data副本，减少内存中重复的数据对象；
  • 使用线程池时可直接通过future.result()获取结果，无需依赖Queue传递，避免队列中残留未清理的对象引用。
• 精准排查内存泄漏：临时开启gc.set_debug(gc.DEBUG_SAVEALL)，查看gc.garbage中的未回收对象，定位循环引用或未释放资源的代码点，排查完成后关闭调试模式。

无内存泄漏的并行实现方案（线程池版） #

替换原update_ping方法，用线程池实现高效并行检测，同时避免内存泄漏：

from concurrent.futures import ThreadPoolExecutor
import gc
import time

def update_ping(self, data, ping_queue):
    # 根据设备数量设置合理线程数，IO密集型任务建议20-50
    with ThreadPoolExecutor(max_workers=40) as executor:
        while not self.stop_threads:
            # 提交所有Ping任务，绑定IP与future对象
            task_map = {}
            for seccion in data:
                for linea in seccion["lineas"]:
                    for equipo in linea["equipos"]:
                        ip = equipo["ip"]
                        future = executor.submit(self.check_online, ip)
                        task_map[future] = ip
            
            # 批量处理任务结果，直接更新原数据
            for future, ip in task_map.items():
                try:
                    is_online = future.result()
                    status_color = "green" if is_online else "red"
                    # 定位对应设备并更新状态
                    for seccion in data:
                        for linea in seccion["lineas"]:
                            for equipo in linea["equipos"]:
                                if equipo["ip"] == ip:
                                    equipo["status_color"] = status_color
                                    break
                except Exception as e:
                    print(f"Error en el hilo {ip}: {str(e)}")
            
            # 直接将更新后的原数据放入队列，避免冗余复制
            ping_queue.put(data)
            print("Ping realizado")
            # 按需触发垃圾回收，避免频繁调用影响性能
            collected = gc.collect()
            print(f"Recolección: {collected}")

def check_online(self, ip):
    check = False
    for _ in range(5):
        data = ping(ip)
        if isinstance(data, float):
            check = True
            break
        else:
            time.sleep(0.5)
    return check

方案优势 #

• 线程池复用线程，彻底避免频繁创建销毁线程的内存开销；
• 直接修改原数据对象，消除冗余数据复制带来的内存占用；
• 用future机制替代队列传递结果，简化逻辑同时减少内存消耗。

额外优化建议 #

• 限制线程池max_workers数量，过多线程会增加上下文切换开销与内存占用，20-50对200台设备足够；
• 检查ping函数内部是否存在未关闭的网络资源或未回收的对象，排除第三方工具的内存泄漏；
• 避免在循环中创建大量临时对象，尽量复用现有数据结构。

内容的提问来源于stack exchange，提问作者Álvaro Blanco Ureña