主进程触发KeyboardInterrupt时如何让子进程抛异常实现优雅清理

实现方案 #

你最后尝试的自定义信号处理思路完全可行，而且不需要轮询，是最优雅的解法，核心逻辑如下：

• 由于Ctrl+C触发的SIGINT信号默认会发送给整个前台进程组，主进程和所有子进程都会收到该信号，不需要主进程额外通过队列/事件传递中断通知
• 子进程在run方法入口就注册自定义的SIGINT信号处理器，收到信号时直接抛出自定义异常，就可以打断任意阶段的执行（包括初始化阶段），统一在顶层try-except块做资源清理
• 你原来需要在初始化每条语句后加检查的问题完全可以避免，异常会直接中断当前执行流，跳到顶层捕获逻辑

适配你的业务场景的代码示例 #

首先自定义中断异常，修改你的sqlExecutor类逻辑：

import multiprocessing
import signal
import logging
import sys
import mysql.connector

# 自定义中断异常
class ProcessInterrupt(Exception):
    pass

class sqlExecutor(multiprocessing.Process):
    def __init__(self, task_queue):
        super().__init__()
        self.task_queue = task_queue
        self.conn = None
        self.cursor = None

    def _sigint_handler(self, signum, frame):
        # 收到SIGINT直接抛自定义异常，打断所有执行流程
        raise ProcessInterrupt
    
    def connect(self):
        self.conn = mysql.connector.connect(
            host="192.168.10.10",
            user="homestead",
            password="xxxx",
            database="xxxx"
        )
        self.cursor = self.conn.cursor() 

    def dispose(self):
        if self.cursor:
            self.cursor.close()
        if self.conn and self.conn.is_connected():
            self.conn.close()

    def init(self):
        # 配置根日志器
        root_logger = logging.getLogger()
        # 你的日志配置逻辑
        # ...
        # 不需要加任何中断检查，收到SIGINT会直接抛异常跳出
        self.connect()
        # 其他初始化逻辑
        # ...
        return True

    def run(self):
        # 子进程启动第一时间注册SIGINT处理器
        signal.signal(signal.SIGINT, self._sigint_handler)
        try:
            self.init()
            while True:
                if not self.conn.is_connected():
                    self.connect()
                row = self.task_queue.get(timeout=5)
                if row is None:
                    self.task_queue.task_done()
                    break
                # 你的数据处理逻辑
                # ...
                self.task_queue.task_done()
        except ProcessInterrupt:
            # 统一处理中断逻辑，不管是初始化阶段还是运行阶段触发
            logging.info("子进程收到中断信号，开始清理资源")
        except Exception as e:
            logging.error(f"子进程执行出错: {str(e)}")
        finally:
            # 所有异常场景最终都走统一资源清理
            self.dispose()
            logging.info("子进程资源清理完成，退出")

主进程逻辑也可以同步简化：

def manage_ctrlC_main(*args):
    logging.info("主进程收到中断信号，等待子进程退出")
    sys.exit()

def main():
    global tasks, sqlDataSaver
    signal.signal(signal.SIGINT, manage_ctrlC_main)
    tasks = multiprocessing.JoinableQueue() 
    sqlDataSaver = sqlExecutor(tasks)
    sqlDataSaver.start() 
    sqlDataSaver.join()

注意事项 #

• 信号处理器是进程级别的，每个子进程独立注册不会互相影响
• 不要在子进程中继承主进程的信号处理逻辑，子进程启动后重新注册自己的处理器即可
• 所有资源清理逻辑统一放在finally块，确保任何退出场景都能执行到

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