如何在Python多进程中实现进程的自定义终止？

作为Python新手，我需要实现以下需求：

• 主进程创建子进程运行另一个文件中的函数
• 用户按下回车键时，通知子进程先执行保存操作，再终止运行（不能直接用multiprocessing.Process.terminate()，因为会直接杀死进程，无法执行保存）

我尝试通过修改全局变量emergency_stop实现，但没有效果，示例代码如下：

主进程代码（文件1） #

import test_global
import msvcrt
import multiprocessing as mp
import time as t

if __name__ == "__main__":
    process = mp.Process(target=test_global.test)
    process.start()
    
    while True:
        # 检测用户按键输入
        if msvcrt.kbhit():
            # 检测回车键
            if msvcrt.getwche() == '\r':
                # 修改全局变量触发停止
                test_global.emergency_stop = True
                print("Emergency Stop Activated")
                print("Clock Stopped")
                break

test_global.py代码 #

import time as t
import datetime as dt

global emergency_stop
emergency_stop = False

def test():
    while True:
        print(f"No input. {dt.datetime.now()}")
        if emergency_stop:
            save()
            print("function stopped")
            break
        t.sleep(1)

def save():
    """
    这里执行保存操作
    """
    pass

我还考虑过管道、队列等方案，但觉得管道需要主进程持续发送空消息，队列需要子进程不断检查，pickle序列化又太复杂，希望找到更合适的方案。

你的全局变量方案无效，是因为multiprocessing创建的子进程拥有独立的内存空间：子进程启动时会复制主进程的全局变量副本，但后续主进程修改的是自己内存中的变量，子进程的副本不会同步更新，所以子进程永远看不到emergency_stop=True的变化。

multiprocessing.Event是专门用于进程间同步的原语，它是进程安全的，主进程调用set()后，子进程可以立即通过is_set()检测到信号，非常适合这种"触发式终止"的场景，代码简洁且性能开销低。

修改后的代码 #

主进程代码（文件1） #

import test_global
import msvcrt
import multiprocessing as mp
import time as t

if __name__ == "__main__":
    # 创建一个事件对象，用于传递终止信号
    stop_event = mp.Event()
    # 将事件对象作为参数传给子进程
    process = mp.Process(target=test_global.test, args=(stop_event,))
    process.start()
    
    while True:
        if msvcrt.kbhit():
            if msvcrt.getwche() == '\r':
                # 设置事件，触发子进程的终止逻辑
                stop_event.set()
                print("Emergency Stop Activated")
                print("Clock Stopped")
                break
    # 等待子进程完成保存操作后再退出主进程
    process.join()

test_global.py代码 #

import time as t
import datetime as dt

def test(stop_event):
    while True:
        print(f"No input. {dt.datetime.now()}")
        # 检测事件是否被触发
        if stop_event.is_set():
            save()
            print("function stopped")
            break
        t.sleep(1)

def save():
    """
    这里执行保存操作
    """
    print("正在执行保存操作...")
    # 替换为你的实际保存逻辑，比如写入文件、数据库等

• 相比全局变量：进程安全，信号传递可靠，不会出现内存副本不同步的问题
• 相比管道/队列：不需要持续发送消息，只需要一次set()操作即可触发终止，代码更简洁
• 性能开销极低：Event是轻量级的同步机制，不会带来额外的性能负担

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