.. _tutorial_step3:

实战三：多任务并发架构——生产者-消费者线程、消息队列与互斥保护
############################################################################

当系统包含周期数据采集、命令交互与报警判断等多个职责时，将所有逻辑塞进 ``main()`` 循环会导致阻塞与耦合。本节我们将 ``smart_node`` 重构为标准的 **RTOS 生产者—消费者架构**，综合运用 :doc:`/develop/kernel/sem`、:doc:`/develop/kernel/mutex` 和 :doc:`/develop/kernel/msgq`。

并发原语选型与架构设计
======================

在动手写代码前，先按以下原则选择内核对象：

.. code-block:: text

   [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`` 实现如下：

.. code-block:: 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``，实测输出如下：

.. code-block:: console

   *** 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

相关参考
========

* :doc:`/develop/kernel/synchronization`
* :doc:`/develop/kernel/sem`
* :doc:`/develop/kernel/mutex`
* :doc:`/develop/kernel/msgq`
* :doc:`/develop/kernel/data_passing`