如何终止通过请求响应队列与主进程通信的子进程？

问题根源 #

主进程仅处理100个请求后就设置event终止子进程，但此时部分子进程已经将请求发送到req_queue中，主进程不会再处理这些未被响应的请求。子进程在执行y = resp_queue.get()时会一直阻塞——因为永远无法等到主进程的响应，即使event已被设置，子进程也没机会进入下一次循环检查event.is_set()。

解决方案 #

1. 处理完所有待响应请求后再终止子进程

在设置event前，先清空req_queue中剩余的请求并发送响应，确保所有子进程的请求都能得到回复。子进程收到响应后会进入下一次循环，此时检查到event已设置就会正常退出。

修改主进程代码：

if __name__ == '__main__':
    event = multiprocessing.Event()
    req_queue = multiprocessing.Queue()
    resp_queues = {}
    processes = {}
    N = 10
    for _ in range(N):  # 启动N个子进程
        resp_queue = multiprocessing.Queue()
        process = multiprocessing.Process(
            target=work, args=(event, req_queue, resp_queue))
        resp_queues[process.name] = resp_queue
        processes[process.name] = process
        process.start()
    for _ in range(100):  # 处理100个请求
        (name, x) = req_queue.get()
        y = x ** 2
        resp_queues[name].put(y)
    
    # 新增：处理队列中剩余的所有请求
    while not req_queue.empty():
        try:
            # 加超时避免意外阻塞
            (name, x) = req_queue.get(timeout=1)
            y = x ** 2
            resp_queues[name].put(y)
        except multiprocessing.queues.Empty:
            break
    
    event.set()  # 通知子进程停止
    for process in processes.values():
        process.join()

2. 给Queue.get()添加超时并检查终止信号

修改子进程的work函数，让resp_queue.get()带有超时时间，超时后立即检查event状态，若已触发终止信号则直接退出循环，避免无限阻塞：

def work(event, req_queue, resp_queue):
    while not event.is_set():
        name = multiprocessing.current_process().name
        x = 3
        req_queue.put((name, x))
        print(name, 'input:', x)
        try:
            # 设置1秒超时，可根据实际情况调整
            y = resp_queue.get(timeout=1)
            print(name, 'output:', y)
        except multiprocessing.queues.Empty:
            # 超时后检查终止信号，若已设置则退出
            if event.is_set():
                break

3. 暴力终止（不推荐，仅作最后手段）

如果上述方法都无法解决，可以直接调用process.terminate()强制终止子进程。但这种方式会直接杀死进程，可能导致资源泄漏（如未关闭的文件、未清理的队列），仅在紧急情况下使用：

event.set()  # 通知子进程停止
for process in processes.values():
    # 先尝试等待5秒，若仍存活则强制终止
    process.join(timeout=5)
    if process.is_alive():
        process.terminate()
        process.join()

内容的提问来源于stack exchange，提问作者Géry Ogam