实战项目一：基于消息队列的日志收集系统

日志收集，听起来好像挺简单？不就是把printf改一改嘛。但真要在嵌入式设备上搞一套靠谱的日志系统，坑多着呢。我记得刚入行那会儿，在一个工业控制器项目里，就因为日志打印把实时任务给拖垮了——串口输出占用了太多CPU时间，导致控制周期超时。从那以后，我就再也不敢小看日志这件事了。

今天这个项目，咱们就用消息队列来搭一套轻量级的日志收集系统。说白了，就是把日志的生产和消费拆开，让日志输出不再阻塞主业务逻辑。

系统整体架构

先看看我们要搭什么。整个系统分三层：

• 日志生产者：各个任务模块，随时产生日志消息
• 消息队列：一个固定大小的环形缓冲区，暂存日志
• 日志消费者：一个后台任务，从队列里取日志，写到串口或SD卡

我习惯把这种模式叫做"异步日志"。你想想看，如果每个任务都直接写串口，那串口锁一争，整个系统都得卡住。用队列一隔，生产者只管往里扔，消费者慢慢往外吐，互不干扰。

消息队列的核心实现

队列用环形缓冲区实现，这是嵌入式里最经典的做法。为什么不用链表？因为动态分配内存容易产生碎片，在资源受限的MCU上是大忌。我见过一个项目，就是因为malloc/free用太多，最后堆碎了，系统莫名其妙死机。

来看代码：

/* 日志消息结构体 */
typedef struct {
    uint32_t timestamp;    /* 系统滴答时间戳 */
    uint8_t  level;        /* 日志级别：DEBUG, INFO, WARN, ERROR */
    uint8_t  module_id;    /* 模块ID */
    char     message[64];  /* 日志内容 */
} log_msg_t;

/* 环形队列结构体 */
typedef struct {
    log_msg_t buffer[LOG_QUEUE_SIZE];
    volatile uint32_t head;  /* 生产者写入位置 */
    volatile uint32_t tail;  /* 消费者读取位置 */
} log_queue_t;

/* 入队函数 */
int log_enqueue(log_queue_t *q, const log_msg_t *msg) {
    uint32_t next_head = (q->head + 1) % LOG_QUEUE_SIZE;
    
    /* 队列满？直接丢弃，不阻塞 */
    if (next_head == q->tail) {
        return -1;  /* 丢日志，但绝不卡任务 */
    }
    
    q->buffer[q->head] = *msg;
    q->head = next_head;
    return 0;
}

/* 出队函数 */
int log_dequeue(log_queue_t *q, log_msg_t *msg) {
    if (q->head == q->tail) {
        return -1;  /* 队列空 */
    }
    
    *msg = q->buffer[q->tail];
    q->tail = (q->tail + 1) % LOG_QUEUE_SIZE;
    return 0;
}

日志级别的设计

日志级别不能太少，也不能太多。我一般用这四级：

级别	枚举值	使用场景
DEBUG	0	开发调试，发布时关掉
INFO	1	正常运行信息，比如"系统启动"
WARN	2	异常但可恢复，比如"通信超时，重试"
ERROR	3	严重错误，比如"传感器无响应"

每个模块在编译时就可以设置自己的日志级别。比如通信模块可以开DEBUG，其他模块只开INFO以上。这样既灵活又省空间。

消费者任务的设计

消费者任务跑在后台，优先级设低一些。它的工作很简单：循环从队列取日志，然后输出。

void log_consumer_task(void *param) {
    log_msg_t msg;
    
    while (1) {
        if (log_dequeue(&g_log_queue, &msg) == 0) {
            /* 格式化输出 */
            char output[128];
            snprintf(output, sizeof(output),
                     "[%lu][%s][mod:%d] %s\n",
                     msg.timestamp,
                     level_to_string(msg.level),
                     msg.module_id,
                     msg.message);
            
            /* 写到串口 */
            uart_send_string(output);
            
            /* 也可以同时写到SD卡，看配置 */
            if (g_log_to_sd) {
                sd_card_append("log.txt", output);
            }
        } else {
            /* 队列空，休眠一会儿 */
            os_delay_ms(10);
        }
    }
}

避坑指南

做这个项目，有几个坑我踩过，你注意一下：

• 队列溢出处理：队列满了怎么办？我的做法是直接丢弃新日志。日志可以丢，但系统不能卡。你想想看，如果因为写日志导致控制任务超时，那才是大问题。
• 中断中写日志：中断服务函数里也可以调用log_enqueue，但要注意——中断里不能调用可能阻塞的函数。我们的入队函数是纯轮询，没有锁，所以安全。
• 时间戳精度：用系统滴答定时器，精度到毫秒就够了。别用RTC，那玩意儿读一次要好几毫秒。
• 日志内容长度：固定64字节够不够？我一般建议128字节。太长浪费内存，太短信息不全。可以根据实际需求调整。

完整的使用示例

最后，看看怎么用这套系统：

/* 初始化 */
log_queue_init(&g_log_queue);
os_task_create(log_consumer_task, "log", 512, NULL, 2);

/* 在任务中写日志 */
void sensor_task(void *param) {
    while (1) {
        int32_t value = read_sensor();
        
        if (value < 0) {
            LOG_ERROR("SENSOR", "读取失败, err=%d", value);
        } else {
            LOG_INFO("SENSOR", "当前值: %d", value);
        }
        
        os_delay_ms(100);
    }
}

这里的LOG_ERROR、LOG_INFO是宏，内部封装了时间戳获取和入队操作。用宏的好处是，可以在DEBUG级别下直接不编译日志代码，省空间。

嗯，这套系统我用了好几年，从STM32到ESP32，从FreeRTOS到裸机，稍微改改就能跑。核心思想就一个：日志是后台的事，别让前台等。你把这个想明白了，日志系统就成功了一大半。

---

公众号:蓝海资料掘金营，微信deep3321