4、日志输出配置：控制台输出、文件输出、网络输出、多进程日志安全

日志输出，说白了就是决定你的日志往哪儿写。

我个人习惯，在ADAS系统开发初期，先把日志输出到控制台。调试方便，一眼能看到。但上了实车或者跑长时间测试，控制台就不够用了。你得写到文件里，甚至通过网络发到远端服务器。

这一节，咱们就把这几种输出方式掰开揉碎讲清楚。还有那个让人头疼的多进程日志安全，我也会分享一些实战经验。

4.1 控制台输出：最直接的调试窗口

控制台输出，是Python logging模块的默认行为。你什么都不配，它默认就往标准错误流（stderr）打印。

但默认格式太丑了。我建议你至少配一下格式，加上时间戳和日志级别。

import logging

# 最简单的控制台输出配置
logging.basicConfig(
    level=logging.DEBUG,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
    datefmt='%Y-%m-%d %H:%M:%S'
)

logger = logging.getLogger('adas.console')
logger.info('控制台输出配置完成')

嗯，这里要注意：basicConfig 只能调用一次。如果你在多个模块里都调它，后面的调用会失效。我遇到过有人在不同文件里重复调用，结果日志级别怎么调都不对。

我的小技巧： 控制台输出适合开发阶段。颜色区分级别？可以用第三方库 colorlog，但别在生产环境用，有性能开销。

4.2 文件输出：持久化存储的关键

文件输出是ADAS系统日志的标配。实车测试一跑就是几个小时，日志量动辄几百MB。控制台根本看不过来，必须写到文件里。

Python的 FileHandler 和 RotatingFileHandler 是主力。

import logging
from logging.handlers import RotatingFileHandler

# 文件输出配置
file_handler = RotatingFileHandler(
    filename='adas_system.log',
    maxBytes=10 * 1024 * 1024,  # 每个文件10MB
    backupCount=5,              # 保留5个备份文件
    encoding='utf-8'
)

file_handler.setLevel(logging.DEBUG)
formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
file_handler.setFormatter(formatter)

logger = logging.getLogger('adas.file')
logger.addHandler(file_handler)
logger.info('文件输出配置完成')

为什么用 RotatingFileHandler？因为日志文件会无限增长。我曾经在测试车上忘了配轮转，结果一个日志文件撑爆了存储卡，导致系统崩溃。嗯，从那以后我再也不敢不配轮转了。

避坑指南： 文件路径一定要用绝对路径，或者确保工作目录正确。我在项目中遇到过，日志文件写到了意想不到的目录，排查了半天才发现是相对路径的问题。

4.3 网络输出：远程监控的利器

ADAS系统有时候部署在无人车上，或者测试场里。你不可能每次都跑到车上去拿日志。这时候，网络输出就派上用场了。

Python的 SocketHandler 可以把日志通过网络发到远程日志服务器。配合 logstash 或 fluentd，还能做集中式日志管理。

import logging
from logging.handlers import SocketHandler

# 网络输出配置
socket_handler = SocketHandler(
    host='192.168.1.100',  # 日志服务器IP
    port=9999              # 端口
)

socket_handler.setLevel(logging.WARNING)  # 网络传输，只发WARNING以上级别
formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
socket_handler.setFormatter(formatter)

logger = logging.getLogger('adas.network')
logger.addHandler(socket_handler)
logger.warning('网络输出配置完成，目标服务器：192.168.1.100:9999')

这里有个坑：网络输出有延迟和丢包风险。我建议只把WARNING和ERROR级别的日志发到网络，DEBUG和INFO级别的日志本地存文件就行。你想想看，如果网络断了，日志发不出去，系统不能因此卡住。

重要提醒： 网络输出一定要加超时和重试机制。否则网络抖动会导致日志线程阻塞，影响主业务逻辑。

4.4 多进程日志安全：最头疼的问题

ADAS系统里，多进程是常态。感知模块一个进程，规划模块一个进程，控制模块一个进程。它们都往同一个日志文件写，问题就来了：日志会乱掉。

Python的logging模块默认不是线程安全的，更不是进程安全的。多进程同时写一个文件，会出现日志交错、内容丢失，甚至文件损坏。

怎么解决？我总结了三种方案：

方案 原理 优点 缺点
方案一：每个进程写独立文件 每个进程创建自己的日志文件，文件名包含进程ID 实现简单，无锁，性能好 日志分散，排查问题需要合并
方案二：使用QueueHandler + QueueListener 子进程把日志放入队列，主进程统一写入文件 日志集中，顺序保证 需要额外进程/线程处理队列
方案三：使用第三方库（如concurrent-log-handler） 基于文件锁或数据库实现进程安全 开箱即用，功能完善 依赖外部库，有性能开销

我个人最推荐方案二。为什么呢？因为它在保证日志顺序的同时，性能损失最小。我在项目中用过方案一，结果有一次系统崩溃，需要从十几个日志文件里拼凑时间线，差点没把我逼疯。

来看看方案二的代码实现：

import logging
import logging.handlers
import multiprocessing
import time

def worker_process(log_queue, process_name):
    """子进程：把日志放入队列"""
    # 使用QueueHandler，日志会自动进入队列
    queue_handler = logging.handlers.QueueHandler(log_queue)
    logger = logging.getLogger(process_name)
    logger.addHandler(queue_handler)
    logger.setLevel(logging.DEBUG)
    
    for i in range(5):
        logger.info(f'{process_name} - 第{i+1}条日志')
        time.sleep(0.1)

def listener_process(log_queue):
    """监听进程：从队列取出日志，写入文件"""
    # 配置文件Handler
    file_handler = logging.FileHandler('merged_adas.log', encoding='utf-8')
    formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
    file_handler.setFormatter(formatter)
    
    # 使用QueueListener
    listener = logging.handlers.QueueListener(log_queue, file_handler)
    listener.start()
    
    # 等待所有子进程结束
    # 实际项目中，这里会一直运行
    time.sleep(3)
    listener.stop()

if __name__ == '__main__':
    log_queue = multiprocessing.Queue(-1)  # 无大小限制
    
    # 启动监听进程
    listener = multiprocessing.Process(target=listener_process, args=(log_queue,))
    listener.start()
    
    # 启动多个工作进程
    processes = []
    for name in ['感知模块', '规划模块', '控制模块']:
        p = multiprocessing.Process(target=worker_process, args=(log_queue, name))
        processes.append(p)
        p.start()
    
    for p in processes:
        p.join()
    
    listener.join()
    print('所有进程日志已安全写入 merged_adas.log')
我的经验： 队列大小要合理设置。太小了会阻塞子进程，太大了会占用内存。我一般设置为10000条，够用。

4.5 综合配置：一个完整的例子

最后，给你一个综合配置的例子。它同时支持控制台、文件、网络输出，并且考虑了多进程安全。

import logging
import logging.handlers
import multiprocessing

def setup_logging(log_queue=None):
    """配置日志系统"""
    logger = logging.getLogger('adas')
    logger.setLevel(logging.DEBUG)
    
    # 1. 控制台输出
    console_handler = logging.StreamHandler()
    console_handler.setLevel(logging.DEBUG)
    console_formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
    console_handler.setFormatter(console_formatter)
    
    # 2. 文件输出（带轮转）
    file_handler = logging.handlers.RotatingFileHandler(
        'adas_system.log', maxBytes=10*1024*1024, backupCount=5, encoding='utf-8'
    )
    file_handler.setLevel(logging.DEBUG)
    file_formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
    file_handler.setFormatter(file_formatter)
    
    # 3. 网络输出（只发WARNING以上）
    network_handler = logging.handlers.SocketHandler('192.168.1.100', 9999)
    network_handler.setLevel(logging.WARNING)
    network_formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
    network_handler.setFormatter(network_formatter)
    
    # 如果是多进程模式，使用QueueHandler
    if log_queue:
        queue_handler = logging.handlers.QueueHandler(log_queue)
        logger.addHandler(queue_handler)
    else:
        logger.addHandler(console_handler)
        logger.addHandler(file_handler)
        logger.addHandler(network_handler)
    
    return logger

# 使用示例
if __name__ == '__main__':
    log_queue = multiprocessing.Queue()
    logger = setup_logging(log_queue)
    logger.info('日志系统初始化完成')

好了，日志输出配置就讲到这里。说白了，控制台输出是开发时的好帮手，文件输出是生产环境的基石，网络输出是远程监控的利器。而多进程日志安全，是ADAS系统绕不开的坎。用QueueHandler + QueueListener的方案，我用了好几年，没出过问题。

下一节，咱们聊聊日志的格式化与过滤。到时候我会分享一些让日志更好看、更好查的技巧。

← 上一章 📖 返回目录 下一章 →