如何利用SIGINT终止multiprocessing.Pool？异步写入JSON数组场景咨询

针对你的场景——用multiprocessing.Pool异步计算数据点、通过回调写入队列，再由单独线程生成JSON数组——处理SIGINT（也就是Ctrl+C）的关键是既要安全终止进程池，又要保证JSON格式的完整性，我给你梳理一套可行的方案：

核心问题分析 #

默认情况下，按下Ctrl+C会给主进程和所有Pool子进程都发送SIGINT信号，容易导致：

• 子进程被强制中断，未完成的任务丢失
• 写入JSON的线程还没处理完队列里的元素，或者没来得及写入闭合的]，最终生成的JSON无效
• 进程池资源没有正确释放，出现僵尸进程

分步解决方案 #

1. 注册SIGINT信号处理函数，控制终止流程 #

我们需要在主进程里捕获SIGINT，替换默认的终止行为，改为优雅收尾：

• 用signal模块注册自定义处理函数
• 在处理函数里设置终止标志，通知进程池和写入线程停止
• 避免在信号处理函数里执行复杂操作（比如IO），因为信号上下文的限制很多

2. 协调进程池、队列和写入线程的收尾逻辑 #

• 对于进程池：如果需要保证已提交的任务都完成，调用pool.close()（停止接受新任务）+ pool.join()（等待所有子进程完成）；如果需要快速终止，用pool.terminate()（立即杀掉所有子进程）
• 对于写入线程：用一个threading.Event作为终止信号，线程循环检查事件，一旦收到终止信号，就处理完队列里的剩余元素，然后写入JSON的闭合括号，最后退出

3. 完整代码示例 #

import multiprocessing
import threading
import signal
import json
import time

# 全局终止标志和事件
terminate_event = threading.Event()
pool = None
json_file = None
data_queue = multiprocessing.Queue()

def compute_data_point(data):
    # 模拟你的数据计算逻辑
    time.sleep(0.1)
    return {"id": data, "value": data * 2}

def callback(result):
    # 回调函数：把计算结果放入队列
    if not terminate_event.is_set():
        data_queue.put(result)

def json_writer():
    # 写入JSON的线程函数
    json_file.write("[")
    first_item = True
    while not terminate_event.is_set() or not data_queue.empty():
        try:
            item = data_queue.get(timeout=0.5)
            if not first_item:
                json_file.write(",")
            json.dump(item, json_file)
            first_item = False
        except multiprocessing.queues.Empty:
            continue
    # 最后写入闭合括号，保证JSON有效
    json_file.write("]")

def sigint_handler(signum, frame):
    print("\n收到终止信号，开始优雅收尾...")
    terminate_event.set()
    # 终止进程池：根据需求选close+join或terminate
    if pool:
        pool.close()  # 不再接受新任务
        pool.join()   # 等待所有已提交任务完成
        # 如果要快速终止，替换成：pool.terminate()

if __name__ == "__main__":
    # 注册SIGINT信号处理
    signal.signal(signal.SIGINT, sigint_handler)

    # 打开JSON文件
    with open("output.json", "w") as f:
        json_file = f
        # 启动写入线程
        writer_thread = threading.Thread(target=json_writer)
        writer_thread.start()

        # 创建进程池
        pool = multiprocessing.Pool()

        # 提交任务
        for i in range(100):
            if terminate_event.is_set():
                break
            pool.apply_async(compute_data_point, args=(i,), callback=callback)

        # 等待终止信号
        while not terminate_event.is_set():
            time.sleep(0.1)

        # 等待写入线程完成
        writer_thread.join()
        print("收尾完成，JSON文件已闭合")

关键注意事项 #

• JSON完整性：一定要确保最后写入]，上面的代码里写入线程会在终止后处理完队列所有元素再写闭合括号，避免生成无效JSON
• 进程池选择：pool.close() + pool.join()会等待所有已提交的任务完成，适合需要完整数据的场景；pool.terminate()会立即终止所有子进程，适合需要快速停止的场景
• 信号处理限制：信号处理函数里只做简单的标志设置，复杂操作交给主流程或线程处理，避免出现不可预期的问题
• 队列空检查：写入线程的循环条件要同时检查终止事件和队列是否为空，保证队列里的所有数据都被处理

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