基于psutil监控目标文件：优化多while循环并行，避免漏检与竞态

问题核心 #

原串行循环的问题在于三个任务依次执行，监控文件的循环会阻塞后续的文件复制和处理流程，导致新生成的目标文件无法及时被检测和处理，甚至漏检。要解决这个问题，我们可以用多线程+线程安全队列来拆分三个独立任务，让它们并行运行，同时避免竞态条件。

优化思路 #

1. 用queue.Queue存储待处理的文件路径，它是线程安全的，无需额外加锁就能在多个线程间安全传递数据。
2. 拆分三个独立线程：
   • 监控线程：定期扫描目标进程的打开文件，将符合扩展名的文件（去重后）加入队列。
   • 复制线程：从队列取出文件，尝试复制到安全目录（处理文件占用的情况）。
   • 处理线程：对安全目录中的文件执行后续操作。

完整代码示例 #

import psutil
import os
import shutil
import queue
import threading
import time

# 配置参数
TARGET_EXTENSIONS = (".xlsx", ".csv", ".docx", ".dat")
SAFE_DIR = "C:\\safe"
TARGET_PID = 12345  # 替换为你的目标进程PID
SCAN_INTERVAL = 1  # 扫描间隔（秒），避免过度占用CPU

# 创建线程安全队列
file_queue = queue.Queue()
processed_files = set()  # 用于去重，避免重复处理同一文件

def monitor_open_files():
    """监控目标进程的打开文件，将符合条件的加入队列"""
    process = psutil.Process(TARGET_PID)
    os.makedirs(SAFE_DIR, exist_ok=True)
    while True:
        try:
            for open_file in process.open_files():
                file_path = open_file.path
                # 检查扩展名是否符合，且未被处理过
                if file_path.lower().endswith(TARGET_EXTENSIONS) and file_path not in processed_files:
                    print(f"检测到目标文件: {file_path}")
                    file_queue.put(file_path)
                    processed_files.add(file_path)
            time.sleep(SCAN_INTERVAL)
        except psutil.NoSuchProcess:
            print("目标进程已退出，监控线程终止")
            break
        except Exception as e:
            print(f"监控线程出错: {str(e)}")
            time.sleep(SCAN_INTERVAL)

def copy_to_safe_dir():
    """从队列取文件，尝试复制到安全目录"""
    while True:
        try:
            file_path = file_queue.get(timeout=5)  # 超时等待，避免无限阻塞
            # 尝试重命名来检查文件是否被占用
            try:
                os.rename(file_path, file_path)
                # 复制到安全目录，保留原文件名
                dest_path = os.path.join(SAFE_DIR, os.path.basename(file_path))
                shutil.copy(file_path, dest_path)
                print(f"文件已复制到安全目录: {dest_path}")
            except OSError:
                # 文件被占用，放回队列稍后重试
                file_queue.put(file_path)
                time.sleep(0.5)
            finally:
                file_queue.task_done()
        except queue.Empty:
            continue
        except Exception as e:
            print(f"复制线程出错: {str(e)}")
            time.sleep(1)

def process_safe_files():
    """处理安全目录中的文件"""
    while True:
        try:
            # 遍历安全目录中的文件，按顺序处理
            for file_name in os.listdir(SAFE_DIR):
                file_path = os.path.join(SAFE_DIR, file_name)
                if os.path.isfile(file_path):
                    print(f"开始处理文件: {file_path}")
                    # 这里替换为你的实际处理逻辑
                    # do_something(file_path)
                    # 处理完成后可以删除或移走文件，避免重复处理
                    # os.remove(file_path)
            time.sleep(2)  # 处理间隔
        except Exception as e:
            print(f"处理线程出错: {str(e)}")
            time.sleep(2)

if __name__ == "__main__":
    # 启动三个线程
    monitor_thread = threading.Thread(target=monitor_open_files, daemon=True)
    copy_thread = threading.Thread(target=copy_to_safe_dir, daemon=True)
    process_thread = threading.Thread(target=process_safe_files, daemon=True)

    monitor_thread.start()
    copy_thread.start()
    process_thread.start()

    # 主线程保持运行，等待所有线程结束
    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        print("程序被用户终止")

关键说明 #

1. 线程安全队列：queue.Queue自动处理线程间的同步，避免多个线程同时操作数据导致的竞态问题。
2. 去重机制：用processed_files集合记录已加入队列的文件，避免同一文件被重复监控。
3. 文件占用处理：复制失败（文件被占用）时将文件放回队列，稍后重试，确保不会丢失文件。
4. 守护线程：设置daemon=True，当主线程退出时，所有子线程会自动终止，避免残留进程。
5. 扫描间隔：添加time.sleep()避免频繁调用psutil.open_files()导致CPU占用过高。

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