如何在Python多进程中获取进程状态并终止指定进程

这里提供几种实用方案，解决你的需求——在process_run执行完成后终止循环运行的process_read进程。

方案1：主进程监控并强制终止（简单直接） #

主进程先等待process_run结束，再主动终止process_read：

import multiprocessing
import time

def run():
    print("run进程启动")
    # 模拟业务逻辑，运行5秒
    time.sleep(5)
    print("run进程执行完毕")

def read_process():
    print("read_process进程启动")
    while True:
        print("读取并打印数据...")
        time.sleep(2)

if __name__ == "__main__":
    process_run = multiprocessing.Process(target=run, name="run")
    process_read = multiprocessing.Process(target=read_process, name="read_process")
    
    process_run.start()
    process_read.start()
    
    # 等待run进程结束
    process_run.join()
    print("检测到run进程已终止，准备停止read_process")
    
    # 检查read_process是否存活，存活则终止
    if process_read.is_alive():
        process_read.terminate()
        # 等待进程完全终止
        process_read.join()
    
    print("所有进程已结束")

说明：

• terminate()会强制终止进程，适用于不需要优雅退出的场景，但可能导致进程内未释放的资源（如文件句柄、网络连接）泄漏。
• join()确保主进程等待目标进程彻底结束后再继续执行。

方案2：用事件（Event）实现优雅退出（推荐） #

通过multiprocessing.Event实现进程间通信，让read_process主动检查退出信号，避免强制终止：

import multiprocessing
import time

def run(exit_event):
    print("run进程启动")
    time.sleep(5)
    print("run进程执行完毕")
    # 设置事件，通知read_process可以退出
    exit_event.set()

def read_process(exit_event):
    print("read_process进程启动")
    while not exit_event.is_set():
        print("读取并打印数据...")
        time.sleep(2)
    print("read_process收到退出信号，正常终止")

if __name__ == "__main__":
    # 创建事件对象，用于进程间信号传递
    exit_event = multiprocessing.Event()
    
    process_run = multiprocessing.Process(target=run, args=(exit_event,), name="run")
    process_read = multiprocessing.Process(target=read_process, args=(exit_event,), name="read_process")
    
    process_run.start()
    process_read.start()
    
    # 等待两个进程全部结束
    process_run.join()
    process_read.join()
    
    print("所有进程已结束")

说明：

• Event是线程/进程安全的信号机制，set()方法标记事件为触发状态，is_set()检查事件是否触发。
• 这种方式让read_process自行退出循环，能正常释放资源，是更安全的做法。

方案3：通过PID检查目标进程状态 #

传递process_run的PID给read_process，让它定期检查目标进程是否存活：

import multiprocessing
import time
import os

def run():
    print("run进程启动")
    time.sleep(5)
    print("run进程执行完毕")

def read_process(run_pid):
    print("read_process进程启动")
    while True:
        try:
            # 向目标PID发送0信号，仅检查进程是否存在（无实际操作）
            os.kill(run_pid, 0)
        except OSError:
            # 进程不存在，退出循环
            print("检测到run进程已终止，read_process开始退出")
            break
        print("读取并打印数据...")
        time.sleep(2)

if __name__ == "__main__":
    process_run = multiprocessing.Process(target=run, name="run")
    process_read = multiprocessing.Process(target=read_process, args=(process_run.pid,), name="read_process")
    
    process_run.start()
    process_read.start()
    
    process_run.join()
    process_read.join()
    
    print("所有进程已结束")

说明：

• 依赖操作系统的PID管理，Windows下os.kill的0信号可能需要替换为其他检查方式（比如用psutil库的pid_exists()方法，需额外安装）。
• 不需要额外的IPC对象，适合轻量级场景。

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