Python中获取函数进程ID并实现进程监控与终止方法咨询

要实现你的需求，核心是通过Python的multiprocessing模块将func1运行在独立子进程中，从而获取其PID并实现监控逻辑。以下是具体实现代码和说明：

完整代码示例 #

import multiprocessing
import time
import os

def func1():
    # 实际业务逻辑：检查ADLS路径下指定数量的文件是否到达
    print(f"func1 启动，进程ID(PID): {os.getpid()}")
    # 模拟文件检查耗时（替换为你的实际ADLS检查代码）
    time.sleep(10)
    print("func1 执行完成：确认所有指定文件已到达")
    return "文件检查结果"

def func2(target_process):
    print(f"func2 启动，开始监控进程 PID: {target_process.pid}")
    
    # 持续检查目标进程是否运行
    while target_process.is_alive():
        print(f"进程 {target_process.pid} 仍在运行，等待中...")
        time.sleep(2)  # 每隔2秒检查一次
    
    # 进程完成后执行文件处理逻辑
    print(f"进程 {target_process.pid} 已结束，开始处理文件")
    # 在这里编写你的文件处理代码（比如读取ADLS文件、数据处理等）
    print("文件处理流程完成")

if __name__ == "__main__":
    # 创建子进程运行func1
    func1_process = multiprocessing.Process(target=func1)
    # 启动子进程
    func1_process.start()
    
    # 调用func2监控该进程并处理后续任务
    func2(func1_process)
    
    # 等待子进程彻底释放资源（可选，因func2已等待进程结束）
    func1_process.join()
    print("所有任务执行完毕")

关键需求实现说明 #

1. 获取func1的进程ID #

通过multiprocessing.Process创建子进程后，直接通过func1_process.pid即可获取func1所在进程的PID。

2. 在func2中检查进程是否运行 #

使用进程对象自带的is_alive()方法，该方法跨平台（支持Windows/Linux/macOS），无需额外依赖，能准确判断进程是否处于运行状态。

3. 进程完成后的处理与终止 #

• 进程正常完成后会自动退出，is_alive()会返回False，此时直接执行后续文件处理逻辑即可。
• 如果需要处理进程超时或异常卡住的情况，可以在监控循环中添加超时强制终止逻辑：

  def func2(target_process, timeout=30):
      start_time = time.time()
      print(f"func2 启动，开始监控进程 PID: {target_process.pid}")
      
      while target_process.is_alive():
          elapsed_time = time.time() - start_time
          if elapsed_time > timeout:
              print(f"进程 {target_process.pid} 超时，强制终止")
              target_process.terminate()  # 强制终止进程
              target_process.join()  # 等待进程资源释放
              break
          print(f"进程 {target_process.pid} 仍在运行，已等待 {int(elapsed_time)} 秒...")
          time.sleep(2)
      
      if not target_process.is_alive():
          print(f"进程 {target_process.pid} 已结束，开始处理文件")
          # 文件处理逻辑
          print("文件处理完成")
  

备选方案：通过PID直接监控（需额外依赖） #

如果必须直接通过PID检查进程状态，可以使用psutil库（需先执行pip install psutil安装）：

import psutil

def is_process_running(pid):
    try:
        process = psutil.Process(pid)
        return process.is_running() and not process.status() == psutil.STATUS_ZOMBIE
    except psutil.NoSuchProcess:
        return False

# 在func2中使用该函数
def func2(pid):
    print(f"func2 启动，开始监控进程 PID: {pid}")
    while is_process_running(pid):
        print(f"进程 {pid} 仍在运行，等待中...")
        time.sleep(2)
    print(f"进程 {pid} 已结束，开始处理文件")
    # 文件处理逻辑

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