导读:本期聚焦于小伙伴创作的《如何在Python中确保父进程被信号杀死后子进程也同步终止》,敬请观看详情,探索知识的价值。以下视频、文章将为您系统阐述其核心内容与价值。如果您觉得《如何在Python中确保父进程被信号杀死后子进程也同步终止》有用,将其分享出去将是对创作者最好的鼓励。

如何在Python中通过信号杀死父进程后确保子进程也终止

在Python多进程编程场景中,我们经常会遇到父进程被信号终止后,子进程变成孤儿进程继续运行的问题。这种情况可能导致资源泄漏、端口占用等异常,因此需要在父进程退出时同步终止所有子进程。本文将介绍几种常见的实现方案,并给出完整的代码示例。

问题背景

当父进程收到终止信号(如SIGTERM、SIGINT)时,默认情况下不会主动通知子进程退出,子进程会被系统的init进程(或systemd等)收养,继续独立运行。要解决这个问题,我们需要在父进程的信号处理函数中,主动终止所有子进程,再退出自身。

方案一:信号处理函数中主动终止子进程

这种方案的核心是给父进程注册信号处理器,当收到终止信号时,先遍历所有子进程并发送终止信号,等待子进程退出后再结束父进程。

import os
import signal
import time
import multiprocessing

# 子进程执行的函数
def child_task():
    print(f"子进程 {os.getpid()} 启动,开始运行")
    try:
        while True:
            print(f"子进程 {os.getpid()} 正在工作...")
            time.sleep(2)
    except KeyboardInterrupt:
        # 子进程收到SIGINT时的处理
        print(f"子进程 {os.getpid()} 收到中断信号,准备退出")
    finally:
        print(f"子进程 {os.getpid()} 已退出")

# 存储所有子进程的列表
child_processes = []

# 信号处理函数
def handle_signal(signum, frame):
    print(f"\n父进程 {os.getpid()} 收到信号 {signum},开始终止所有子进程")
    # 遍历所有子进程,发送SIGTERM信号
    for p in child_processes:
        if p.is_alive():
            print(f"向子进程 {p.pid} 发送终止信号")
            p.terminate()  # 发送SIGTERM信号
    # 等待所有子进程退出
    for p in child_processes:
        p.join()
        print(f"子进程 {p.pid} 已完全退出")
    print(f"父进程 {os.getpid()} 退出")
    # 退出父进程,返回信号对应的退出码
    os._exit(128 + signum)

if __name__ == "__main__":
    # 注册信号处理器,处理SIGTERM和SIGINT信号
    signal.signal(signal.SIGTERM, handle_signal)
    signal.signal(signal.SIGINT, handle_signal)
    
    # 创建3个子进程
    for i in range(3):
        p = multiprocessing.Process(target=child_task)
        p.start()
        child_processes.append(p)
        print(f"启动子进程 {p.pid}")
    
    print(f"父进程 {os.getpid()} 启动完成,等待信号...")
    try:
        # 父进程保持运行,等待信号
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        handle_signal(signal.SIGINT, None)

上面的代码中,我们首先定义了一个全局的子进程列表,用来存储所有启动的子进程实例。然后注册了SIGTERM和SIGINT信号的处理函数,当父进程收到这两个信号时,会先遍历子进程列表,对每个存活的子进程调用terminate()方法发送SIGTERM信号,再通过join()等待子进程完全退出,最后父进程调用os._exit()退出。需要注意的是,这里使用os._exit()而不是sys.exit(),因为sys.exit()会抛出SystemExit异常,可能被信号处理器之外的逻辑捕获,导致退出不彻底。

方案二:使用进程组统一发送信号

如果子进程数量较多,或者子进程还会继续创建孙进程,逐个终止子进程的方式可能不够高效。此时可以将父进程和所有子进程放入同一个进程组,发送信号给整个进程组,实现批量终止。

import os
import signal
import time
import multiprocessing

def child_task():
    print(f"子进程 {os.getpid()} 属于进程组 {os.getpgrp()},开始运行")
    try:
        while True:
            print(f"子进程 {os.getpid()} 正在工作...")
            time.sleep(2)
    finally:
        print(f"子进程 {os.getpid()} 已退出")

def handle_signal(signum, frame):
    print(f"\n父进程 {os.getpid()} 收到信号 {signum},向进程组发送终止信号")
    # 获取当前进程的进程组ID
    pgid = os.getpgrp()
    # 向整个进程组发送SIGTERM信号,注意使用os.killpg,第一个参数是进程组ID,第二个是信号
    # 这里用-pgid可以发送信号给进程组,包括父进程自身
    os.killpg(pgid, signal.SIGTERM)
    # 等待子进程退出(可选,因为信号已经发送给所有进程)
    time.sleep(1)
    print(f"父进程 {os.getpid()} 退出")
    os._exit(128 + signum)

if __name__ == "__main__":
    # 注册信号处理器
    signal.signal(signal.SIGTERM, handle_signal)
    signal.signal(signal.SIGINT, handle_signal)
    
    # 创建子进程
    processes = []
    for i in range(3):
        p = multiprocessing.Process(target=child_task)
        p.start()
        processes.append(p)
    
    print(f"父进程 {os.getpid()},进程组ID {os.getpgrp()},启动完成")
    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        handle_signal(signal.SIGINT, None)

这种方案的优势在于不需要维护子进程列表,只要所有子进程和父进程属于同一个进程组,发送信号给进程组就可以一次性终止所有相关进程。不过需要注意,如果子进程中调用了os.setpgrp()修改了自己的进程组,那么这种方案就会失效,因为子进程已经脱离了原来的进程组。

方案三:使用prctl设置父进程退出时终止子进程(Linux专用)

在Linux系统下,可以通过prctl系统调用设置子进程的选项,当父进程退出时,子进程自动收到SIGKILL信号终止。这种方式不需要父进程主动处理信号,由系统内核自动完成。

import os
import signal
import time
import multiprocessing
import ctypes

# 定义prctl的常数,PR_SET_PDEATHSIG表示设置父进程死亡时发送的信号
PR_SET_PDEATHSIG = 1
SIGKILL = 9

def child_task():
    # 加载libc库,调用prctl设置父进程退出时发送SIGKILL给子进程
    libc = ctypes.CDLL("libc.so.6")
    ret = libc.prctl(PR_SET_PDEATHSIG, SIGKILL, 0, 0, 0)
    if ret != 0:
        print(f"子进程 {os.getpid()} 设置PDEATHSIG失败")
    else:
        print(f"子进程 {os.getpid()} 已设置父进程退出时自动终止")
    
    try:
        while True:
            print(f"子进程 {os.getpid()} 正在工作...")
            time.sleep(2)
    finally:
        print(f"子进程 {os.getpid()} 已退出")

if __name__ == "__main__":
    # 创建子进程
    processes = []
    for i in range(3):
        p = multiprocessing.Process(target=child_task)
        p.start()
        processes.append(p)
    
    print(f"父进程 {os.getpid()} 启动完成,按Ctrl+C或发送SIGTERM测试")
    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        print("\n父进程收到中断信号,准备退出")
        # 父进程退出后,子进程会自动收到SIGKILL终止
        os._exit(0)

这种方式是Linux系统特有的,跨平台性较差,但实现起来更简单,不需要父进程维护子进程列表或者处理信号。不过需要注意,如果子进程在设置PR_SET_PDEATHSIG之前父进程就已经退出,那么子进程不会收到终止信号,所以设置操作要尽可能在子进程启动后尽早执行。

不同方案的选择建议

  • 如果需要在所有类Unix系统和Windows系统上运行,优先选择方案一,手动维护子进程列表并在信号处理函数中终止子进程,兼容性最好。
  • 如果只在Linux系统运行,且子进程不会修改自己的进程组,方案二的进程组方式代码更简洁,适合子进程数量多、层级深的场景。
  • 如果希望子进程在父进程意外退出(比如被SIGKILL强制杀死)时也能自动终止,方案三的prctl方式更合适,因为SIGKILL信号无法被捕获,父进程无法主动处理,而prctl由内核触发,不受信号捕获限制。

注意事项

首先,SIGKILL信号(信号值为9)是无法被捕获和忽略的,所以如果父进程是被SIGKILL杀死的,方案一和方案二的信号处理函数不会执行,此时只有方案三的prctl方式可以保证子进程终止。其次,子进程中如果有需要清理的资源,最好在子进程中也注册对应的信号处理函数,在收到终止信号时完成资源释放,避免数据丢失或者资源泄漏。另外,使用multiprocessing.Processterminate()方法时,子进程可能不会执行finally块中的代码,所以重要的清理逻辑最好放在子进程的信号处理函数中。

Python多进程信号终止孤儿进程进程组prctl 本作品最后修改时间:2026-05-23 22:07:25

免责声明:已尽一切努力确保本网站所含信息的准确性。网站部分内容来源于网络或由用户自行发表,内容观点不代表本站立场。本站是个人网站免费分享,内容仅供个人学习、研究或参考使用,如内容中引用了第三方作品,其版权归原作者所有。若内容触犯了您的权益,请联系我们进行处理。
内容垂直聚焦
专注技术核心技术栏目,确保每篇文章深度聚焦于实用技能。从代码技巧到架构设计,为用户提供无干扰的纯技术知识沉淀,精准满足专业提升需求。
知识结构清晰
覆盖从开发到部署的全链路。前端、网络、数据库、服务器、建站、系统层层递进,构建清晰学习路径,帮助用户系统化掌握网站开发与运维所需的核心技术栈。
深度技术解析
拒绝泛泛而谈,深入技术细节与实践难点。无论是数据库优化还是服务器配置,均结合真实场景与代码示例进行剖析,致力于提供可直接应用于工作的解决方案。
专业领域覆盖
精准对应开发生命周期。从前端界面到后端逻辑,从数据库操作到服务器运维,形成完整闭环,一站式满足全栈工程师和运维人员的技术需求。
即学即用高效
内容强调实操性,步骤清晰、代码完整。用户可根据教程直接复现和应用于自身项目,显著缩短从学习到实践的距离,快速解决开发中的具体问题。
持续更新保障
专注既定技术方向进行长期、稳定的内容输出。确保各栏目技术文章持续更新迭代,紧跟主流技术发展趋势,为用户提供经久不衰的学习价值。