如何在Python中实现随目标函数运行的CPU使用率监控循环？

核心思路是把XML解析和CPU监控拆成两个独立线程：主线程执行解析逻辑，子线程负责循环监控CPU，直到解析完成后自动终止监控。

具体实现方案 #

用threading.Event作为线程间的终止信号：解析函数开始前触发事件，解析结束后清除事件，监控线程循环检查事件状态，一旦事件被清除就停止监控。

完整代码示例 #

import threading
import time
import psutil
import xml.etree.ElementTree as ET

# 初始化终止信号，默认未触发
monitor_running = threading.Event()

def monitor_cpu(output_file):
    with open(output_file, 'w', encoding='utf-8') as f:
        f.write("时间戳,CPU使用率(%)\n")
        # 只要信号触发，就持续监控
        while monitor_running.is_set():
            timestamp = time.strftime("%Y-%m-%d %H:%M:%S.%f")[:-3]
            # 每0.1秒采样一次CPU使用率
            cpu_usage = psutil.cpu_percent(interval=0.1)
            f.write(f"{timestamp},{cpu_usage}\n")
            f.flush()  # 实时写入文件，避免缓存积压

def parse_xml(xml_path):
    start_time = time.time()
    # 触发信号，通知监控线程开始工作
    monitor_running.set()
    
    # 执行XML解析逻辑
    tree = ET.parse(xml_path)
    root = tree.getroot()
    
    # 解析完成，清除信号，终止监控
    monitor_running.clear()
    end_time = time.time()
    
    # 输出解析结果和耗时
    print(f"解析耗时: {end_time - start_time:.2f}秒")
    print("XML树结构:")
    ET.dump(root)
    return tree

if __name__ == "__main__":
    # 启动监控线程
    cpu_thread = threading.Thread(target=monitor_cpu, args=("cpu_monitor.log",))
    cpu_thread.start()
    
    # 执行XML解析
    parse_xml("test.xml")
    
    # 等待监控线程完全结束
    cpu_thread.join()

关键细节说明 #

• monitor_running是线程通信的核心：set()表示解析正在进行，监控要继续；clear()表示解析完成，监控停止。
• psutil.cpu_percent(interval=0.1)：interval参数控制采样间隔，0.1秒的间隔既能保证数据实时性，又不会占用过多系统资源。
• f.flush()：强制把缓存内容写入文件，避免程序结束后才批量写入，确保监控数据完整。

替代方案：通过线程存活状态判断 #

如果不想用事件信号，也可以直接检查解析线程的存活状态，监控线程只要发现解析线程还在运行就继续工作：

def monitor_cpu(output_file, target_thread):
    with open(output_file, 'w', encoding='utf-8') as f:
        f.write("时间戳,CPU使用率(%)\n")
        while target_thread.is_alive():
            timestamp = time.strftime("%Y-%m-%d %H:%M:%S.%f")[:-3]
            cpu_usage = psutil.cpu_percent(interval=0.1)
            f.write(f"{timestamp},{cpu_usage}\n")
            f.flush()

if __name__ == "__main__":
    # 启动解析线程
    parse_thread = threading.Thread(target=parse_xml, args=("test.xml",))
    parse_thread.start()
    
    # 启动监控线程，传入解析线程对象
    cpu_thread = threading.Thread(target=monitor_cpu, args=("cpu_monitor.log", parse_thread))
    cpu_thread.start()
    
    parse_thread.join()
    cpu_thread.join()

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