如何在多进程并发场景下从另一个函数中退出目标函数？

首先纠正原代码里的两个关键错误：

• 创建Process时，target参数要传函数对象（比如target=A），而不是函数调用结果（target=A()），后者会直接在主进程执行函数，而非子进程
• 多进程拥有独立的内存空间，普通变量condition无法在进程间共享，必须用multiprocessing模块提供的进程间通信工具

下面给出两种可行的实现方式：

方法一：用Event让进程A主动退出（推荐） #

这种方式更稳妥，让A进程自己检测退出信号，避免强制终止带来的资源泄漏问题。

import multiprocessing
import time

def A(stop_event):
    while not stop_event.is_set():
        # 替换成A进程的实际业务逻辑
        print("A进程运行中...")
        time.sleep(1)
    print("A进程收到退出信号，主动结束")

def B(stop_event):
    while True:
        # 模拟触发退出条件的逻辑，这里用3秒等待代替实际判断
        time.sleep(3)
        print("B进程触发退出条件，发送停止信号")
        stop_event.set()
        # 发送信号后B进程可根据需求选择退出或继续运行
        break

def main():
    # 创建进程间共享的Event对象，用于传递退出信号
    stop_event = multiprocessing.Event()
    
    # 给子进程传递参数时，用args元组传入
    p1 = multiprocessing.Process(target=A, args=(stop_event,))
    p2 = multiprocessing.Process(target=B, args=(stop_event,))
    
    p1.start()
    p2.start()
    
    p1.join()
    p2.join()
    print("所有进程执行完毕")

if __name__ == '__main__':
    main()

核心说明：

• multiprocessing.Event()是进程安全的同步对象，set()方法会标记退出信号，is_set()用于检查信号状态
• A进程循环检测信号，一旦信号被设置就退出循环，主动结束进程
• B进程满足条件时调用stop_event.set()，给A进程发送退出指令

方法二：让B进程强制终止A进程 #

如果A进程无法主动检测信号（比如业务逻辑阻塞），可以用terminate()方法强制终止，但这种方式可能导致资源未正确释放，非必要不推荐。

import multiprocessing
import time

def A():
    while True:
        print("A进程运行中...")
        time.sleep(1)

def B(p1):
    while True:
        # 模拟触发退出条件
        time.sleep(3)
        print("B进程触发退出条件，强制终止A进程")
        p1.terminate()
        p1.join()  # 等待A进程完全终止
        break

def main():
    p1 = multiprocessing.Process(target=A)
    p2 = multiprocessing.Process(target=B, args=(p1,))
    
    p1.start()
    p2.start()
    
    p2.join()
    print("所有进程执行完毕")

if __name__ == '__main__':
    main()

核心说明：

• B进程接收A进程的对象作为参数，调用terminate()强制结束A进程
• 强制终止会直接杀死进程，可能导致A进程中打开的文件、网络连接等资源无法正常关闭，需谨慎使用

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