嵌入式软件架构:前后台系统 vs RTOS、任务划分原则、消息队列与状态机设计

好,咱们今天聊点实在的。嵌入式软件架构,说白了就是你怎么组织你的代码,让它跑得稳、改得动、出了问题还能快速定位。我在分拣机项目里见过太多因为架构没选好,最后整个产线停摆的案例。嗯,这节课我就把几个核心问题掰开揉碎了讲清楚。

前后台系统 vs RTOS:怎么选?

先说说前后台系统。这玩意儿也叫超级循环,就是main函数里一个死循环,轮询各种标志位。我早期做小家电时,一个电饭煲的控制器,前后台系统完全够用。为什么?因为任务简单,就几个按键、一个温度检测、一个加热控制,逻辑不复杂。

但分拣机不一样。你想想看,它要同时处理传感器信号、电机控制、通信协议、故障检测……这些任务对实时性的要求还不一样。比如急停信号,必须在几毫秒内响应,否则可能出安全事故。这时候前后台系统就扛不住了。

我个人习惯是这么判断的:

  • 任务数少于5个,且没有硬实时要求 → 前后台系统,简单可靠
  • 任务数超过5个,或者有毫秒级响应需求 → 上RTOS
  • 需要任务间通信、资源共享 → 必须RTOS

核心区别:前后台系统是“你做完我再做”,RTOS是“谁急谁先做”。

我在项目中遇到过一件事。有个同事非要在前后台系统里硬塞一个复杂的通信协议栈,结果主循环里一个延时函数就把整个系统卡死了。后来换成RTOS,把通信任务单独放一个线程,问题就解决了。说白了,工具选对了,事半功倍。

任务划分原则:别把鸡蛋放一个篮子里

任务怎么分?这是个艺术活。我见过有人把整个分拣逻辑写成一个任务,结果改一个功能要动半套代码。也见过有人分得太细,任务间通信比干活还多。

我总结了几条原则,你记一下:

  1. 按功能边界划分:传感器采集一个任务,电机控制一个任务,通信处理一个任务。每个任务只干一件事。
  2. 按实时性要求划分:急停、故障检测这种高优先级任务单独拎出来。数据显示这种低优先级任务可以合并。
  3. 按资源访问划分:共享同一个外设的操作尽量放一个任务里,减少锁的竞争。
  4. 任务粒度要适中:一个任务执行时间别超过系统滴答周期的1/3。我一般控制在5毫秒以内。

小技巧:画个任务依赖图。如果两个任务之间频繁通信,考虑合并成一个。如果某个任务代码超过500行,考虑拆开。

举个例子。分拣机的核心流程:

  • 任务A:读取光电传感器状态(2ms周期)
  • 任务B:根据传感器数据计算分拣动作(5ms周期)
  • 任务C:控制气缸/电机执行分拣(10ms周期)
  • 任务D:通过串口上报状态给上位机(100ms周期)

你看,每个任务周期不同,优先级也不同。A和B必须高优先级,C可以中等,D最低。这样设计,就算D偶尔卡一下,也不影响分拣动作。

消息队列:任务间的“快递员”

任务之间怎么传数据?用全局变量?我劝你慎重。全局变量最大的问题是:你不知道它什么时候被谁改了。调试起来跟破案一样。

消息队列就好多了。它像一个带锁的邮箱,任务A往里扔消息,任务B从里面取。谁都不会干扰谁。

我常用的消息队列设计:

// 定义消息结构体
typedef struct {
    uint8_t msg_id;      // 消息ID
    uint8_t data_len;    // 数据长度
    uint8_t data[64];    // 数据内容
} Msg_t;

// 创建消息队列
osMessageQDef(msg_queue, 10, Msg_t);
osMessageQId msg_queue_id = osMessageCreate(osMessageQ(msg_queue), NULL);

// 任务A发送消息
Msg_t msg;
msg.msg_id = 0x01;
msg.data_len = 2;
msg.data[0] = sensor_value;
msg.data[1] = checksum;
osMessagePut(msg_queue_id, &msg, 0);

// 任务B接收消息
osEvent event = osMessageGet(msg_queue_id, osWaitForever);
if (event.status == osEventMessage) {
    Msg_t *recv_msg = (Msg_t *)event.value.p;
    // 根据msg_id处理不同消息
}

注意:消息队列长度要合理。太短容易丢消息,太长浪费内存。我一般根据任务的最大积压量再加20%余量。比如任务B每秒处理10条消息,任务A每秒最多发15条,那队列长度设20就够。

我曾经在一个项目里吃过亏。消息队列长度设了5,结果上位机突发大量指令,队列满了直接丢消息。分拣机就开始乱动,把好好的零件扔进了废品箱。后来改成动态队列,问题才解决。嗯,教训深刻。

状态机设计:让代码“有章可循”

分拣机的控制逻辑,说白了就是一个状态机。什么状态做什么事,清清楚楚。我最怕看到那种if-else套了七八层的代码,改一个条件要翻半天。

状态机设计我推荐用查表法。结构清晰,扩展方便。

// 定义状态枚举
typedef enum {
    STATE_IDLE,       // 空闲
    STATE_DETECT,     // 检测
    STATE_SORT,       // 分拣
    STATE_ERROR,      // 错误
    STATE_MAX
} State_t;

// 定义事件枚举
typedef enum {
    EVT_NONE,
    EVT_SENSOR_TRIG,  // 传感器触发
    EVT_SORT_DONE,    // 分拣完成
    EVT_ERROR,        // 错误发生
    EVT_RESET         // 复位
} Event_t;

// 状态转移表
typedef struct {
    State_t current_state;
    Event_t event;
    State_t next_state;
    void (*action)(void);  // 动作函数指针
} StateTable_t;

// 状态转移表实例
StateTable_t state_table[] = {
    {STATE_IDLE,   EVT_SENSOR_TRIG, STATE_DETECT, start_detect},
    {STATE_DETECT, EVT_SORT_DONE,   STATE_SORT,   start_sort},
    {STATE_SORT,   EVT_SORT_DONE,   STATE_IDLE,   stop_sort},
    {STATE_ERROR,  EVT_RESET,       STATE_IDLE,   reset_system},
    // ... 其他状态转移
};

// 状态机执行函数
void state_machine_run(Event_t event) {
    for (int i = 0; i < sizeof(state_table)/sizeof(StateTable_t); i++) {
        if (state_table[i].current_state == current_state &&
            state_table[i].event == event) {
            // 执行动作
            if (state_table[i].action) {
                state_table[i].action();
            }
            // 状态转移
            current_state = state_table[i].next_state;
            break;
        }
    }
}

你看,这样设计,新增一个状态或者事件,只需要在表里加一行。代码可读性也高,一眼就能看出整个系统的行为逻辑。

经验之谈:状态机里一定要加一个“兜底”状态。比如所有未定义的状态转移,统一进ERROR状态。这样就算出了意外,系统也能安全停下来,而不是乱跑。

我有个习惯,每个状态机都配一个状态日志。每次状态变化,把当前状态、触发事件、时间戳都记录下来。产线出问题时,翻翻日志就知道当时发生了什么。这招帮我省了不少排查时间。

好了,这节课的内容就这些。前后台还是RTOS,任务怎么分,消息队列怎么用,状态机怎么写——这几个问题想清楚了,分拣机的软件架构就稳了。下一节我们聊聊具体的调试技巧,到时候见。