实战三：多任务并发架构——生产者-消费者线程、消息队列与互斥保护
当系统包含周期数据采集、命令交互与报警判断等多个职责时，将所有逻辑塞进 main() 循环会导致阻塞与耦合。本节我们将 smart_node 重构为标准的 RTOS 生产者—消费者架构，综合运用 信号量、互斥量 和 消息队列。
并发原语选型与架构设计
在动手写代码前，先按以下原则选择内核对象：
[Shell 命令: node trigger / 中断]
│
│ 1. 事件通知（唤醒线程，不传结构体）──► k_sem (sample_trig_sem)
▼
┌──────────────────────┐ 2. 传递固定大小采样结构体 ┌──────────────────────┐
│ sampler_thread │ ───────────────────────────► │ processor_thread │
│ (生产者，优先级 6) │ k_msgq (sample_msgq) │ (消费者，优先级 5) │
└──────────────────────┘ └──────────┬───────────┘
│
│ 3. 保护共享配置读写
│ k_mutex (config_mutex)
▼
┌──────────────────────┐
│ g_temp_threshold │
└──────────────────────┘
为什么生产者用
k_sem_take(&sample_trig_sem, K_MSEC(500))代替k_msleep(500)？ 这样既能实现“平时每 500ms 超时自动采集一次（返回-EAGAIN）”，又能在用户执行node trigger调用k_sem_give()时**零延迟立即唤醒采集**（返回0）！为什么把消费者优先级（5）设得比生产者（6）更高？ 在 Zephyr 中数字越小优先级越高。当
sampler_thread调用k_msgq_put()放入数据的瞬间，阻塞在k_msgq_get()上的高优先级processor_thread会立即抢占 CPU 完成告警处理。
完整代码实现（src/main.c）
prj.conf 仅需启用 CONFIG_LOG=y 与 CONFIG_SHELL=y。src/main.c 实现如下：
#include <zephyr/kernel.h>
#include <zephyr/logging/log.h>
#include <zephyr/shell/shell.h>
#include <stdlib.h>
LOG_MODULE_REGISTER(smart_node_ipc, LOG_LEVEL_INF);
struct sensor_sample {
int64_t timestamp_ms;
int32_t temp_c;
uint32_t seq;
};
/* 消息队列：最多缓存 8 条采样消息，4 字节对齐 */
K_MSGQ_DEFINE(sample_msgq, sizeof(struct sensor_sample), 8, 4);
/* 二值信号量：初始计数 0，最大计数 1，用于立即唤醒采集线程 */
K_SEM_DEFINE(sample_trig_sem, 0, 1);
/* 互斥量：保护多线程共享的报警阈值 */
K_MUTEX_DEFINE(config_mutex);
static int32_t g_temp_threshold = 30;
/* 生产者线程：周期采集或被信号量立即唤醒 */
static void sampler_thread_entry(void *p1, void *p2, void *p3)
{
ARG_UNUSED(p1);
ARG_UNUSED(p2);
ARG_UNUSED(p3);
static const int32_t temps[] = { 25, 28, 33, 29, 36 };
uint32_t seq = 0;
while (seq < 5) {
int ret = k_sem_take(&sample_trig_sem, K_MSEC(500));
bool manual = (ret == 0);
struct sensor_sample item = {
.timestamp_ms = k_uptime_get(),
.temp_c = temps[seq % ARRAY_SIZE(temps)],
.seq = seq++,
};
LOG_INF("[Producer] Sample #%u (%s): temp=%d C",
item.seq, manual ? "MANUAL" : "PERIODIC", item.temp_c);
if (k_msgq_put(&sample_msgq, &item, K_NO_WAIT) != 0) {
LOG_WRN("Message queue full, dropping sample #%u", item.seq);
}
}
}
/* 消费者线程：阻塞读取消息队列并执行阈值判断 */
static void processor_thread_entry(void *p1, void *p2, void *p3)
{
ARG_UNUSED(p1);
ARG_UNUSED(p2);
ARG_UNUSED(p3);
struct sensor_sample item;
while (1) {
if (k_msgq_get(&sample_msgq, &item, K_FOREVER) == 0) {
k_mutex_lock(&config_mutex, K_FOREVER);
int32_t th = g_temp_threshold;
k_mutex_unlock(&config_mutex);
if (item.temp_c > th) {
LOG_WRN("[Consumer] ALARM on #%u: %d C > threshold %d C",
item.seq, item.temp_c, th);
} else {
LOG_INF("[Consumer] Normal #%u: %d C <= threshold %d C",
item.seq, item.temp_c, th);
}
}
}
}
K_THREAD_DEFINE(sampler_tid, 2048, sampler_thread_entry, NULL, NULL, NULL, 6, 0, 0);
K_THREAD_DEFINE(processor_tid, 2048, processor_thread_entry, NULL, NULL, NULL, 5, 0, 0);
static int cmd_node_trigger(const struct shell *sh, size_t argc, char **argv)
{
ARG_UNUSED(argc);
ARG_UNUSED(argv);
k_sem_give(&sample_trig_sem);
shell_print(sh, "Immediate sample triggered via semaphore");
return 0;
}
SHELL_STATIC_SUBCMD_SET_CREATE(sub_node,
SHELL_CMD_ARG(trigger, NULL, "Trigger immediate sample", cmd_node_trigger, 1, 0),
SHELL_SUBCMD_SET_END
);
SHELL_CMD_REGISTER(node, &sub_node, "Smart Node IPC commands", NULL);
int main(void)
{
LOG_INF("Producer-Consumer IPC started");
return 0;
}
在 qemu_cortex_a53 上验证并发时序
编译运行并在串口输入 node trigger，实测输出如下：
*** Booting Zephyr OS build 321e39a45766 ***
[00:00:00.010,000] <inf> smart_node_ipc: Producer-Consumer IPC started
[00:00:00.520,000] <inf> smart_node_ipc: [Producer] Sample #0 (PERIODIC): temp=25 C
[00:00:00.520,000] <inf> smart_node_ipc: [Consumer] Normal #0: 25 C <= threshold 30 C
uart:~$ node trigger
Immediate sample triggered via semaphore
[00:00:00.980,000] <inf> smart_node_ipc: [Producer] Sample #1 (MANUAL): temp=28 C
[00:00:00.980,000] <inf> smart_node_ipc: [Consumer] Normal #1: 28 C <= threshold 30 C
[00:00:01.490,000] <inf> smart_node_ipc: [Producer] Sample #2 (PERIODIC): temp=33 C
[00:00:01.490,000] <wrn> smart_node_ipc: [Consumer] ALARM on #2: 33 C > threshold 30 C
[00:00:02.000,000] <inf> smart_node_ipc: [Producer] Sample #3 (PERIODIC): temp=29 C
[00:00:02.000,000] <inf> smart_node_ipc: [Consumer] Normal #3: 29 C <= threshold 30 C
[00:00:02.510,000] <inf> smart_node_ipc: [Producer] Sample #4 (PERIODIC): temp=36 C
[00:00:02.510,000] <wrn> smart_node_ipc: [Consumer] ALARM on #4: 36 C > threshold 30 C