使用join（）后如何在执行过程中终止进程？

Question

我想在执行过程中杀死某个进程或所有进程。我的示例代码如下。我怎样才能做到这一点？在这里，我想根据当前时间终止进程。如果时间可以被 2 整除，我想杀死进程，否则不会。

import time
from multiprocessing import Process


def runTests(a, b):
    time.sleep(10)
    return a + b


def main(kill_processes):
    print(kill_processes)
    processes = []

    for i in range(3):
        print(i)
        proc = Process(target=runTests, args=(2, 4,))
        processes.append(proc)
        proc.start()

    for proc in processes:
        proc.join()
        if kill_processes:
            print("killing")
            proc.terminate()


if __name__ == "__main__":
    if round(time.time()) % 2 == 0:
        main(True)
    else:
        main(False)

Answer 1

这说明了如何以类似于我建议您查看的问题multiprocessing in stoppable multithreading 的答案中的方式使用多个线程来做到这一点。

基本上所有的join()调用都是在单独的线程中完成的，所以它们不会阻塞主进程中的主线程——这使它有可能杀死它们。

import threading
import time
from multiprocessing import Process


def runTests(a, b):
    time.sleep(10)
    return a + b

def create_process(lock, i):
    proc = Process(target=runTests, args=(2, 4,))
    print(f'{proc.name} created')
    proc.start()
    with lock:
        processes.append(proc)
    proc.join()


def main(kill_processes):
    global processes
    N = 3
    lock = threading.RLock()
    processes = []

    print(f'main({kill_processes=})')

    for i in range(N):
        thread = threading.Thread(target=create_process, args=(lock, i))
        thread.start()

    while True:  # Wait for all processes to have been created.
        with lock:
            if len(processes) == N:
                break
            else:
                time.sleep(.001)

    if kill_processes:
        print("Killing the processes")
        for proc in processes:
            proc.terminate()
            print(f'process {proc} terminated')



if __name__ == "__main__":
    main(True)

#    if round(time.time()) % 2 == 0:
#        main(True)
#    else:
#        main(False)

Answer 2

您可以使用multiprocessing.Event 向您的子进程发出终止条件信号。不要在主进程中加入子进程。而是让主进程和子进程在自己的循环中运行。检查主循环中的终止条件，并使用multiprocessing.Event 向子进程发出信号。

Event 对象作为参数传递给子进程。子进程不断检查是否设置了事件，如果设置了则停止其工作。主循环检查终止条件并在满足条件时设置Event（在下面的示例中，主循环等待Ctrl+c）。

import multiprocessing as mp
import os
import time


def do_work(a, b, stop_event):
    while not stop_event.is_set():
        try:
            time.sleep(2)
            print(f"worker {os.getpid()}: working ...", a + b)
            a += 1
            b += 1
        except KeyboardInterrupt:
            print(f"worker {os.getpid()}: received SIGINT. ignore.")
            pass

    print(f"worker {os.getpid()}: stop_event is set. exit.")


if __name__ == "__main__":
    stop_event = mp.Event()
    procs = []
    for i in range(3):
        # p = mp.Process(target=do_work, args=(1, 2, stop_event), daemon=True)
        p = mp.Process(target=do_work, args=(1, 2, stop_event))
        p.start()
        procs.append(p)

    while True:
        try:
            print("main: waiting for termination signal")
            time.sleep(1)
        except KeyboardInterrupt:
            print("main: received termination signal")            
            stop_event.set()
            
            # wait for the processes to stop
            for p in procs:
                p.join()

            for p in procs:
                print(f"worker {p.pid} is terminated: {not p.is_alive()}")
            
            # exit the main loop
            break
    
    print("main: bye")

Answer 3

如果要根据时间终止，请使用join 的timeout 参数。一种方法是设置停止时间，当每个进程加入时，使用剩余时间作为超时时间。

import time
from multiprocessing import Process

def runTests(a, b):
    time.sleep(10)
    return a + b

def main(kill_processes):
    print(kill_processes)
    processes = []

    end = time.time() + 10 # wait 10 seconds

    for i in range(3):
        print(i)
        proc = Process(target=runTests, args=(2, 4,))
        processes.append(proc)
        proc.start()

    for proc in processes:
        if kill_processes:
            delta = end - time.time()
        else:
            delta = None
        proc.join(delta)
        if kill_processes:
            print("killing")
            proc.terminate()
            proc.join(1)
            if proc.is_alive():
                proc.kill()

if __name__ == "__main__":
    if round(time.time()) % 2 == 0:
        main(True)
    else:
        main(False)