Zephyr RTOS中多线程接收UART新数据并访问的实现方案咨询

针对你遇到的高优先级线程独占数据、数据覆盖的问题，以下两个可行方案可满足需求：

方案1：k_event广播 + 带引用计数的缓存节点 #

核心思路：每次UART接收数据后生成独立缓存副本，用k_event唤醒所有订阅线程，通过引用计数确保数据在所有线程处理完成前不被释放。

实现步骤 #

1. 定义缓存节点与全局资源

#include <zephyr/kernel.h>
#include <zephyr/sys/slist.h>
#include <stdatomic.h>

// 自定义UART接收数据结构体
struct uart_rx_data {
    uint8_t buf[64];
    size_t len;
};

// 带引用计数的缓存节点
struct uart_data_node {
    struct uart_rx_data data;
    atomic_t ref_count; // 记录未处理数据的线程数
    struct slist_node node;
};

// 预先分配内存池管理缓存节点，避免动态内存碎片
K_MEM_POOL_DEFINE(uart_data_pool, sizeof(struct uart_data_node),
                  sizeof(struct uart_data_node) * 8, 1, 0);

// 全局广播事件：通知所有线程新数据到来
struct k_event uart_data_event;
// 订阅线程链表：统计需要接收通知的线程数量
struct slist_head subscriber_list;
// 互斥锁：保护链表与缓存节点的并发操作
struct k_mutex uart_mutex;

2. 订阅线程注册
   每个需要接收数据的线程初始化时完成注册：

void uart_subscribe(void) {
    k_mutex_lock(&uart_mutex, K_FOREVER);
    slist_add_tail(&subscriber_list, (struct slist_node *)k_current_get());
    k_mutex_unlock(&uart_mutex);
}

3. UART接收线程处理逻辑
   收到数据后生成缓存节点，设置引用计数为订阅线程数，触发广播事件：

void uart_rx_thread(void *p1, void *p2, void *p3) {
    struct uart_rx_data raw_data;
    struct uart_data_node *node;
    size_t subscriber_count;

    while (1) {
        // 替换为实际UART读取逻辑
        if (uart_receive(&raw_data) == 0) {
            k_mutex_lock(&uart_mutex, K_FOREVER);

            // 从内存池申请节点
            node = k_mem_pool_alloc(&uart_data_pool, K_FOREVER);
            // 复制数据到独立节点
            memcpy(&node->data, &raw_data, sizeof(struct uart_rx_data));
            // 初始化引用计数为订阅线程总数
            subscriber_count = slist_size(&subscriber_list);
            atomic_set(&node->ref_count, subscriber_count);

            k_mutex_unlock(&uart_mutex);

            // 广播事件，传递缓存节点指针
            k_event_set(&uart_data_event, 0x1, (void *)node);
        }
    }
}

4. 订阅线程处理逻辑
   等待事件触发，处理数据后更新引用计数，计数为0时释放节点：

void subscriber_thread(void *p1, void *p2, void *p3) {
    struct k_event event;
    struct uart_data_node *node;
    int ret;

    // 注册为订阅者
    uart_subscribe();

    while (1) {
        ret = k_event_wait(&uart_data_event, 0x1, &event, K_FOREVER);
        if (ret == 0) {
            node = (struct uart_data_node *)event.user_data;

            // 替换为你的业务处理逻辑
            process_uart_data(&node->data);

            // 引用计数减1，为0时释放节点回内存池
            if (atomic_dec(&node->ref_count) == 0) {
                k_mem_pool_free(&uart_data_pool, node);
            }
        }
    }
}

方案优势 #

• 独立缓存节点彻底避免数据覆盖问题，引用计数确保数据安全释放
• k_event可同时唤醒所有等待线程，无论优先级，每个线程都能获取专属数据副本
• 内存池管理节点，规避动态内存分配的碎片化风险

方案2：共享环形缓存 + 线程独立读取偏移 #

核心思路：UART线程将数据写入环形缓存，每个订阅线程维护自己的读取偏移，用k_event广播通知，线程根据自身偏移读取新数据。

实现步骤 #

1. 定义全局缓存与线程上下文

#include <zephyr/kernel.h>
#include <zephyr/sys/slist.h>

#define UART_BUF_SIZE 512

// 环形缓存结构体
struct uart_ring_buf {
    uint8_t buf[UART_BUF_SIZE];
    size_t write_idx;
    size_t read_idx;
    struct k_mutex lock;
};

// 订阅线程上下文：记录自身读取偏移
struct subscriber_ctx {
    size_t read_idx;
    struct slist_node node;
};

// 全局环形缓存
struct uart_ring_buf uart_ring;
// 全局广播事件
struct k_event uart_data_event;
// 订阅线程上下文链表
struct slist_head subscriber_ctx_list;
struct k_mutex ctx_mutex;

2. 订阅线程初始化
   每个线程初始化自身读取偏移并完成注册：

void subscriber_init(struct subscriber_ctx *ctx) {
    k_mutex_lock(&ctx_mutex, K_FOREVER);
    k_mutex_lock(&uart_ring.lock, K_FOREVER);
    // 初始读取偏移与缓存写入偏移同步
    ctx->read_idx = uart_ring.write_idx;
    k_mutex_unlock(&uart_ring.lock);
    slist_add_tail(&subscriber_ctx_list, &ctx->node);
    k_mutex_unlock(&ctx_mutex);
}

3. UART接收线程写入缓存

void uart_rx_thread(void *p1, void *p2, void *p3) {
    uint8_t rx_buf[64];
    size_t rx_len;

    while (1) {
        // 替换为实际UART读取逻辑
        rx_len = uart_read(rx_buf, sizeof(rx_buf));
        if (rx_len > 0) {
            k_mutex_lock(&uart_ring.lock, K_FOREVER);

            // 环形缓存写入逻辑
            size_t copy_len = MIN(rx_len, UART_BUF_SIZE - uart_ring.write_idx);
            memcpy(&uart_ring.buf[uart_ring.write_idx], rx_buf, copy_len);
            if (rx_len > copy_len) {
                memcpy(uart_ring.buf, &rx_buf[copy_len], rx_len - copy_len);
            }
            uart_ring.write_idx = (uart_ring.write_idx + rx_len) % UART_BUF_SIZE;

            k_mutex_unlock(&uart_ring.lock);

            // 广播通知所有线程
            k_event_set(&uart_data_event, 0x1, NULL);
        }
    }
}

4. 订阅线程读取数据

void subscriber_thread(void *p1, void *p2, void *p3) {
    struct subscriber_ctx ctx = {0};
    uint8_t process_buf[64];
    size_t process_len;

    // 初始化订阅上下文
    subscriber_init(&ctx);

    while (1) {
        k_event_wait(&uart_data_event, 0x1, NULL, K_FOREVER);
        k_mutex_lock(&uart_ring.lock, K_FOREVER);

        // 计算可用数据长度
        if (uart_ring.write_idx >= ctx.read_idx) {
            process_len = uart_ring.write_idx - ctx.read_idx;
        } else {
            process_len = UART_BUF_SIZE - ctx.read_idx + uart_ring.write_idx;
        }

        if (process_len > 0) {
            // 读取数据到本地缓冲区
            size_t copy_len = MIN(process_len, sizeof(process_buf));
            if (ctx.read_idx + copy_len <= UART_BUF_SIZE) {
                memcpy(process_buf, &uart_ring.buf[ctx.read_idx], copy_len);
            } else {
                size_t first_part = UART_BUF_SIZE - ctx.read_idx;
                memcpy(process_buf, &uart_ring.buf[ctx.read_idx], first_part);
                memcpy(&process_buf[first_part], uart_ring.buf, copy_len - first_part);
            }
            // 更新自身读取偏移
            ctx.read_idx = (ctx.read_idx + copy_len) % UART_BUF_SIZE;

            // 替换为你的业务处理逻辑
            process_uart_data(process_buf, copy_len);
        }

        k_mutex_unlock(&uart_ring.lock);
    }
}

方案优势 #

• 无需为每个数据创建副本，内存占用更低
• 每个线程独立管理读取进度，不会互相干扰
• 环形缓存适配持续的UART数据流场景

内容的提问来源于stack exchange，提问作者Illya Kuzmych