You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

如何用Python实现跨Windows/Linux的进程启停事件通知式检测

实现跨平台的进程启动/关闭事件通知(非轮询方式)

问题背景

需要用Python编写程序,检测特定进程的启停并触发预设操作(如进程关闭时备份文件夹),要求采用事件驱动机制(类似inotify)替代定期轮询,避免psutil轮询在Windows多进程环境下的高CPU占用问题,同时兼容Windows和Linux系统。

此前采用psutil轮询的代码存在性能问题:每5秒遍历所有进程一次,Windows环境下遍历耗时超1秒,会占满单个CPU核心:

import psutil
import time

while True:
  for process in psutil.process_iter():
    if process.name().lower() == PROCESSNAME:
      # 进程存在时执行的操作
      ...

  time.sleep(5)

解决方案

Linux平台:利用inotify监控/proc目录

Linux的/proc文件系统会为每个运行进程创建以PID命名的目录,通过监控/proc目录的CREATE和DELETE事件,可直接感知进程启停,再通过PID校验进程名实现精准过滤。

使用pyinotify库实现:

import pyinotify
import psutil

PROCESS_NAME = "target_process"  # 替换为目标进程名

class ProcessEventProcessor(pyinotify.ProcessEvent):
    def process_IN_CREATE(self, event):
        # 过滤/proc下的数字PID目录
        if event.name.isdigit():
            try:
                pid = int(event.name)
                proc = psutil.Process(pid)
                if proc.name().lower() == PROCESS_NAME.lower():
                    print(f"进程启动: {PROCESS_NAME} (PID: {pid})")
                    # 执行进程启动时的操作
                    # backup_folder()
            except (psutil.NoSuchProcess, psutil.AccessDenied):
                pass

    def process_IN_DELETE(self, event):
        if event.name.isdigit():
            pid = int(event.name)
            # 可结合缓存的目标PID列表,确认是目标进程后触发操作
            print(f"进程关闭: PID {pid}")
            # 执行进程关闭时的操作
            # backup_folder()

if __name__ == "__main__":
    wm = pyinotify.WatchManager()
    mask = pyinotify.IN_CREATE | pyinotify.IN_DELETE
    notifier = pyinotify.Notifier(wm, ProcessEventProcessor())
    wm.add_watch('/proc', mask, rec=False)
    notifier.loop()

Windows平台:利用WMI事件订阅

Windows可通过WMI(Windows Management Instrumentation)订阅进程创建/删除事件,直接接收系统推送的事件通知,无需轮询。

使用wmi库实现:

import wmi
import psutil

PROCESS_NAME = "target_process.exe"  # 替换为目标进程名

def on_process_created(event):
    pid = int(event.ProcessId)
    try:
        proc = psutil.Process(pid)
        if proc.name().lower() == PROCESS_NAME.lower():
            print(f"进程启动: {PROCESS_NAME} (PID: {pid})")
            # 执行进程启动时的操作
    except (psutil.NoSuchProcess, psutil.AccessDenied):
        pass

def on_process_deleted(event):
    pid = int(event.ProcessId)
    # 结合缓存的目标PID列表,确认是目标进程后触发操作
    print(f"进程关闭: PID {pid}")
    # 执行进程关闭时的操作

if __name__ == "__main__":
    c = wmi.WMI()
    # 订阅进程创建事件
    c.Win32_Process.watch_for("creation", callback=on_process_created)
    # 订阅进程删除事件
    c.Win32_Process.watch_for("deletion", callback=on_process_deleted)
    # 保持程序运行
    while True:
        pass

跨平台整合代码

通过platform模块自动识别系统,适配对应的监控逻辑:

import platform
import psutil

PROCESS_NAME = "target_process"  # 根据系统调整后缀(如Windows加.exe)

def handle_process_start(pid):
    print(f"进程启动: {PROCESS_NAME} (PID: {pid})")
    # 替换为你的启动操作
    # example: backup_folder()

def handle_process_stop(pid):
    print(f"进程关闭: PID {pid}")
    # 替换为你的关闭操作
    # example: backup_folder()

if __name__ == "__main__":
    system = platform.system()
    if system == "Linux":
        import pyinotify

        class ProcessEvent(pyinotify.ProcessEvent):
            def process_IN_CREATE(self, event):
                if event.name.isdigit():
                    try:
                        pid = int(event.name)
                        proc = psutil.Process(pid)
                        if proc.name().lower() == PROCESS_NAME.lower():
                            handle_process_start(pid)
                    except (psutil.NoSuchProcess, psutil.AccessDenied):
                        pass

            def process_IN_DELETE(self, event):
                if event.name.isdigit():
                    pid = int(event.name)
                    # 可添加PID缓存逻辑,避免误触发
                    handle_process_stop(pid)

        wm = pyinotify.WatchManager()
        mask = pyinotify.IN_CREATE | pyinotify.IN_DELETE
        notifier = pyinotify.Notifier(wm, ProcessEvent())
        wm.add_watch('/proc', mask, rec=False)
        notifier.loop()
    elif system == "Windows":
        import wmi

        def on_create(event):
            pid = int(event.ProcessId)
            try:
                proc = psutil.Process(pid)
                if proc.name().lower() == PROCESS_NAME.lower():
                    handle_process_start(pid)
            except (psutil.NoSuchProcess, psutil.AccessDenied):
                pass

        def on_delete(event):
            pid = int(event.ProcessId)
            handle_process_stop(pid)

        c = wmi.WMI()
        c.Win32_Process.watch_for("creation", callback=on_create)
        c.Win32_Process.watch_for("deletion", callback=on_delete)
        while True:
            pass
    else:
        print(f"不支持的系统: {system}")

注意事项

  • Linux下监控/proc会产生大量无关事件,建议维护目标PID缓存列表,避免误触发。
  • Windows下WMI事件存在轻微延迟,但CPU占用远低于轮询;需确保Python进程有足够权限监控系统进程。
  • 依赖安装:Linux执行pip install pyinotify psutil;Windows执行pip install wmi psutil。

内容的提问来源于stack exchange,提问作者Daniel Carrasco Marín

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.07.26 05:45:01