Python多进程错误未写入日志及异常捕获优化咨询

一、全面的异常捕获策略 #

多进程环境下，单一特定异常捕获（比如仅KeyError）无法覆盖所有可能的错误，建议采用分层捕获方式，从业务特定异常到通用异常再到系统级异常，确保任务队列不会因未捕获的异常中断：

需覆盖的核心异常类型 #

• 业务逻辑异常：KeyError, IndexError, ValueError, TypeError, NameError
• IO/文件操作异常：FileNotFoundError, PermissionError, IOError, OSError
• 多进程特有异常：multiprocessing.ProcessError, multiprocessing.TimeoutError, queue.Empty, queue.Full, BrokenPipeError
• 通用异常：Exception（捕获所有非系统级的运行时异常，避免遗漏）
• 系统级异常：BaseException下的KeyboardInterrupt, SystemExit, MemoryError（按需处理，比如KeyboardInterrupt需优雅关闭进程池）

捕获最佳实践 #

不要使用裸except:（会捕获包括SystemExit在内的所有异常，导致无法正常退出），而是按优先级分层捕获：

try:
    # 你的数据分析任务代码
    value = data['missing_key']
except KeyError as e:
    # 处理特定业务异常
    logger.error(f"Missing key: {str(e)}", exc_info=True)
except (FileNotFoundError, PermissionError) as e:
    # 批量处理IO相关异常
    logger.error(f"File operation failed: {str(e)}", exc_info=True)
except Exception as e:
    # 兜底捕获所有非系统级异常
    logger.error(f"Unexpected runtime error: {str(e)}", exc_info=True)
except BaseException as e:
    # 处理系统级异常，保证优雅退出
    logger.critical(f"System error occurred: {str(e)}", exc_info=True)
    sys.exit(1)

注：exc_info=True会记录完整的异常栈，便于定位问题

二、多进程日志写入失败的解决方案 #

多进程直接共享logging文件句柄会导致资源竞争，出现日志丢失、乱码或写入失败的问题，推荐两种可靠的解决方式：

方式1：使用进程安全的日志处理器（Python 3.9+） #

使用RotatingFileHandler或TimedRotatingFileHandler，开启lock=True参数保证进程安全，每个子进程独立初始化日志：

import multiprocessing
import logging
from logging.handlers import RotatingFileHandler
import sys

def setup_logger():
    """子进程独立初始化日志处理器"""
    logger = logging.getLogger('data_analysis')
    logger.setLevel(logging.ERROR)
    # 避免重复添加handler
    if not logger.handlers:
        handler = RotatingFileHandler(
            'data_analysis.log',
            maxBytes=10*1024*1024,  # 单日志文件最大10MB
            backupCount=5,  # 保留5个备份
            encoding='utf-8',
            delay=True,
            lock=True  # 开启进程锁，保证多进程写入安全
        )
        formatter = logging.Formatter('%(asctime)s - %(processName)s - %(levelname)s - %(message)s')
        handler.setFormatter(formatter)
        logger.addHandler(handler)
    return logger

def process_data(data):
    logger = setup_logger()
    try:
        value = data['missing_key']
        print(value)
    except KeyError as e:
        logger.error(f"KeyError occurred: {str(e)}", exc_info=True)
    except Exception as e:
        logger.error(f"Unexpected error: {str(e)}", exc_info=True)
    except BaseException as e:
        logger.critical(f"System error: {str(e)}", exc_info=True)
        sys.exit(1)

if __name__ == '__main__':
    data_list = [{'key1': 'val1'}, {'key2': 'val2'}]
    with multiprocessing.Pool(2) as pool:
        pool.map(process_data, data_list)

方式2：日志队列模式（高并发场景推荐） #

主进程启动一个独立的日志消费者进程，所有子进程将日志消息发送到队列，由消费者统一写入文件，彻底避免进程竞争：

import multiprocessing
import logging
import sys
from queue import Empty

def logger_process(queue):
    """日志消费者进程，负责写入文件"""
    handler = RotatingFileHandler(
        'data_analysis.log',
        maxBytes=10*1024*1024,
        backupCount=5,
        encoding='utf-8'
    )
    formatter = logging.Formatter('%(asctime)s - %(processName)s - %(levelname)s - %(message)s')
    handler.setFormatter(formatter)
    logger = logging.getLogger('queue_logger')
    logger.addHandler(handler)
    logger.setLevel(logging.ERROR)

    while True:
        try:
            record = queue.get(timeout=1)
            if record is None:  # 收到退出信号
                break
            logger.handle(record)
        except Empty:
            continue

def process_data(data, queue):
    """子进程，将日志消息发送到队列"""
    logger = logging.getLogger('queue_logger')
    try:
        value = data['missing_key']
        print(value)
    except KeyError as e:
        logger.error(f"KeyError occurred: {str(e)}", exc_info=True)
    except Exception as e:
        logger.error(f"Unexpected error: {str(e)}", exc_info=True)
    except BaseException as e:
        logger.critical(f"System error: {str(e)}", exc_info=True)
        sys.exit(1)

if __name__ == '__main__':
    # 创建日志队列
    log_queue = multiprocessing.Queue()
    # 启动日志消费者进程
    log_process = multiprocessing.Process(target=logger_process, args=(log_queue,))
    log_process.start()

    # 配置日志发送到队列
    queue_handler = logging.handlers.QueueHandler(log_queue)
    root_logger = logging.getLogger()
    root_logger.addHandler(queue_handler)
    root_logger.setLevel(logging.ERROR)

    # 启动数据处理进程池
    data_list = [{'key1': 'val1'}, {'key2': 'val2'}]
    with multiprocessing.Pool(2) as pool:
        pool.map(lambda x: process_data(x, log_queue), data_list)

    # 发送退出信号，关闭日志进程
    log_queue.put(None)
    log_process.join()

关键注意事项 #

• 子进程中不要直接传递主进程的logger对象，多进程间无法共享线程安全的对象，需独立初始化或通过队列传递日志消息
• 始终记录完整的异常栈（exc_info=True），否则仅靠错误信息难以定位多进程中的问题
• 避免在多进程中使用logging.basicConfig，会导致多个进程重复创建文件句柄

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