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
相关产品推荐
相关产品推荐

