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

Python进程检测与终止:如何在信号处理器运行前终止main进程?

Python进程控制与信号处理资源冲突解决方案

1. 检测并终止进程

方法一:使用psutil库(推荐,跨平台)

先安装依赖:pip install psutil

检测进程(按名称匹配):

import psutil

def find_process_by_name(process_name):
    for proc in psutil.process_iter(['name']):
        if proc.info['name'] == process_name:
            return proc.pid
    return None

终止进程(先尝试优雅终止,失败则强制终止):

import os
import signal
import time
import psutil

def kill_process(pid):
    try:
        # 发送SIGTERM信号优雅终止进程
        os.kill(pid, signal.SIGTERM)
        # 等待2秒确认进程是否退出
        time.sleep(2)
        if psutil.pid_exists(pid):
            # 仍存活则发送SIGKILL强制终止
            os.kill(pid, signal.SIGKILL)
        print(f"进程{pid}已终止")
    except OSError as e:
        print(f"终止进程失败: {e}")

方法二:使用系统命令(平台特定)

Linux/macOS环境:

import subprocess
subprocess.run(["pkill", "-f", "your_script.py"], check=True)

Windows环境:

import subprocess
subprocess.run(["taskkill", "/F", "/IM", "python.exe", "/FI", "WINDOWTITLE eq your_script"], check=True)

2. 识别运行main()的进程并在signal_handler执行前终止其逻辑

如果main()是当前进程的主函数,直接用os.getpid()获取当前进程PID。要在信号处理前让main()停止执行并释放资源,可用事件通知的方式:

import signal
import time
import os
from threading import Event

# 全局事件,用于通知main退出
exit_event = Event()

def main():
    print("Main函数执行中")
    print("等待信号...")
    
    # 注册信号处理函数
    signal.signal(signal.SIGHUP, signal_handler)
    
    # 用事件等待代替time.sleep,可被信号中断
    exit_event.wait(timeout=10)
    
    # 手动释放公共资源
    cleanup_shared_resources()
    print("Main函数已退出")

def cleanup_shared_resources():
    # 在这里释放Function2使用的资源,比如关闭文件、释放锁等
    pass

def signal_handler(sig, frame):
    # 先通知main停止执行,等待资源释放
    exit_event.set()
    time.sleep(0.5)  # 给main预留资源释放时间
    
    # 执行信号处理逻辑
    print("开始处理SIGHUP信号")
    # 原代码的日志和文件删除逻辑放在这里

如果要识别其他运行main()的进程,可通过进程命令行判断:

import psutil

def find_main_process(script_name):
    target_cmd = f"python {script_name}"
    for proc in psutil.process_iter(['pid', 'cmdline']):
        if proc.info['cmdline'] and ' '.join(proc.info['cmdline']) == target_cmd:
            return proc.pid
    return None

3. 解决main与signal_handler共用资源的IO错误

核心是避免公共资源的并发访问——Python信号处理在主线程触发,会打断main的执行流程,直接访问同一资源会导致IO冲突,以下是两种解决思路:

方案一:用锁保护公共资源

给Function2的公共资源操作加锁,确保同一时间只有一个逻辑访问:

from threading import Lock

# 全局锁保护公共资源
resource_lock = Lock()

def Function2():
    with resource_lock:
        # 公共资源操作,比如文件读写
        with open("shared_file.txt", "r+") as f:
            content = f.read()
            f.seek(0)
            f.write(content + "updated")
            f.truncate()

# main中调用Function2
def main():
    # ...
    with resource_lock:
        Function2()
    # ...

# 信号处理中调用Function2
def signal_handler(sig, frame):
    # ...
    with resource_lock:
        Function2()
    # ...

方案二:让信号处理只做通知,main主动清理资源

信号处理函数尽量简短,只设置退出标志,让main线程完成当前操作后主动释放资源:

import signal
import time

exit_flag = False

def main():
    global exit_flag
    print("Main执行中")
    signal.signal(signal.SIGHUP, signal_handler)
    
    while not exit_flag:
        # 执行main的逻辑,比如调用Function2
        Function2()
        time.sleep(1)
    
    cleanup_shared_resources()
    print("Main退出")

def signal_handler(sig, frame):
    global exit_flag
    exit_flag = True
    print("收到信号,将在main完成当前操作后退出")

def cleanup_shared_resources():
    # 释放资源逻辑
    pass

修正原代码的路径错误

原代码中os.path.join(os.getcwd(),"/Del_folder/index.txt")会生成无效路径(第二个参数以/开头会忽略前面的路径),正确写法是:

os.path.join(os.getcwd(), "Del_folder", "index.txt")

内容的提问来源于stack exchange,提问作者Adarsh Udupi

相关产品推荐
方舟 Agent Plan

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

最近更新时间:2026.08.17 05:35:21