ESP-IDF+vscode开发ESP32第十五讲——队列、流缓冲区、环形缓冲区

目录

前言

一、三种机制简介

1. 队列 (Queue)

2. 流缓冲区 (Stream Buffer)

3. 环形缓冲区 (Ring Buffer)

4. 核心区别对比表

5. 选型指南

二、流缓冲区 (Stream Buffer) 使用

2.1 uart_m.c

2.2 uart_m.h

2.3 main.c

2.4 代码解释

2.5 结果展示

三、环形缓冲区 (Ring Buffer)使用

前言

在 ESP32开发中，队列（Queue）、流缓冲区（Stream Buffer）和环形缓冲区（Ring Buffer）是三种最常用的任务间通信和数据缓冲机制。

虽然它们都能实现“一个任务/中断发数据，另一个任务收数据”，但它们的设计初衷、数据组织方式、并发限制和性能开销有着本质的区别。本章就这三种机制进行一个讲解和使用。

一、三种机制简介

1. 队列 (Queue)

• 来源：FreeRTOS 原生组件。
• 数据组织：固定大小的离散数据项（Item）。创建时必须指定每个 Item 的大小（如 4 字节的int，或 32 字节的struct）。它像是一列火车，每节车厢大小固定，按先进先出（FIFO）顺序排队。
• 并发支持：支持多读多写（内部有互斥锁保护，线程安全）。
• 内存机制：发生内存拷贝。发送时把数据memcpy进队列，接收时把数据memcpy出来。
• 典型应用场景：
  • 传递传感器离散采样值（如温度、湿度）。
  • 传递控制命令、按键事件、状态标志。
  • 传递结构体（如果结构体很大，通常传递结构体的指针，但需自行管理指针指向的内存生命周期）。

2. 流缓冲区 (Stream Buffer)

• 来源：FreeRTOS 原生组件（从 v10.0.0 引入）。
• 数据组织：连续的字节流（Byte Stream）。没有“Item”边界的概念，你可以一次写 10 字节，分两次每次读 5 字节。
• 并发支持：严格的单读单写（Single Reader, Single Writer）。为了追求极致的性能，它去掉了内部的互斥锁。这意味着只能有一个任务往里写，一个任务从里读，绝不能多任务并发读写。
• 内存机制：发生内存拷贝。
• 典型应用场景：
  • 中断 (ISR) 到任务的数据传递（因为无锁，在中断中执行极快）。
  • UART 串口接收不定长的连续数据流。
  • ADC 连续采样产生的数据流。

3. 环形缓冲区 (Ring Buffer)

• 来源：ESP-IDF 特有组件（非 FreeRTOS 原生，专为 ESP32 优化）。
• 数据组织：支持变长数据项或纯字节流。它提供了三种类型：
  • RINGBUF_TYPE_NOSPLIT：Item 必须存放在连续的物理内存中（最常用）。、
  • RINGBUF_TYPE_ALLOWSPLIT：允许 Item 在缓冲区尾部截断并绕回头部（节省空间但读取复杂）。
  • RINGBUF_TYPE_BYTEBUF：纯字节流（类似 Stream Buffer）。
• 并发支持：支持多读多写（内部有锁保护）。
• 内存机制：支持零拷贝（Zero-Copy）。这是它最大的杀手锏！通过SendAcquire和Receive，你可以直接获取 Ring Buffer 内部内存的指针，直接在里面读写数据，完全省去了memcpy的开销。
• 典型应用场景：
  • 音视频流处理（如 I2S 麦克风采集、MP3 解码数据传递）。
  • DMA 数据缓冲（让 DMA 直接把数据写到 Ring Buffer 内部）。
  • 网络数据包（Wi-Fi/蓝牙）的封包与解包。
  • 任何数据量大、频率高，且对 CPU 拷贝开销敏感的场景。

4. 核心区别对比表

5. 选型指南

场景 1：传递控制信号、事件、小结构体

选择 Queue

• 理由：数据量小（几个字节到几十个字节），拷贝开销可以忽略不计。Queue 支持多读多写，API 最丰富，支持阻塞等待，是传递“离散状态”最稳妥的选择。

场景 2：传递连续的字节流（如串口数据），且只有“一发一收”

选择 Stream Buffer

• 理由：没有 Item 大小的限制，非常适合处理不定长的字节流。由于它没有锁，在 UART 接收中断（ISR）中调用xStreamBufferSendFromISR的速度极快，不会导致中断阻塞。

场景 3：传递大块数据（如音频、图像、大数组），且对性能要求极高

选择 Ring Buffer

• 理由：当你要传递 1KB 甚至更大的数据块时，Queue 和 Stream Buffer 的memcpy会消耗大量 CPU 时间。Ring Buffer 的零拷贝特性允许你直接操作底层内存，或者直接把 Ring Buffer 的内部指针交给 DMA 控制器，从而实现极高的数据吞吐量。

总结一句话

• Queue 是用来传 “消息/事件” 的（稳妥、通用）。
• Stream Buffer 是用来传 “字节流” 的（轻量、无锁、单发单收）。
• Ring Buffer 是用来传 “大数据块” 的（零拷贝、高性能、ESP32 专属）。

说了以上这么多，大家对这三种机制应该有了一个大致的印象，但是如果没有使用开发过，很难去体会它的特点。下面就分别对流缓冲区和环形缓冲区进行使用示例，因为队列属于基础，相信大家都会使用，这里就不再重复了

二、流缓冲区 (Stream Buffer) 使用

我们用《ESP-IDF+vscode开发ESP32第三讲——UART_vscode 开发esp32-CSDN博客》该文的工程来实现流缓冲区的作用。代码如下：

2.1 uart_m.c

#include <stdio.h> #include "uart_m.h" #include "driver/gpio.h" #include "driver/uart.h" #include "freertos//stream_buffer.h" static const char *TAG = "UART"; QueueHandle_t uart_intr_handle= NULL; StreamBufferHandle_t uart_stream_buffer = NULL; void uart_intr_callback(void *arg); void stream_buffer_info(void *arg); void uart_init(void) { uart_stream_buffer = xStreamBufferCreate(1024, 100); uart_config_t uart_config = { .baud_rate = 115200, .data_bits = UART_DATA_8_BITS, .parity = UART_PARITY_DISABLE, .stop_bits = UART_STOP_BITS_1, .flow_ctrl = UART_HW_FLOWCTRL_DISABLE, .source_clk = UART_SCLK_DEFAULT, }; ESP_ERROR_CHECK(uart_driver_install(uart_port, uart_rx_buffer, uart_tx_buffer, 5, &uart_intr_handle, 0)); ESP_ERROR_CHECK(uart_param_config(uart_port, &uart_config)); ESP_ERROR_CHECK(uart_set_pin(uart_port, uart_tx_pin, uart_rx_pin, -1, -1, -1, -1)); ESP_ERROR_CHECK(uart_set_mode(uart_port, UART_MODE_UART)); ESP_LOGI(TAG, uart_is_driver_installed(uart_port)? "UART driver is installed" : "UART driver is not installed"); //中断配置 uart_intr_config_t intr_config = { .intr_enable_mask = uart_intr, .rx_timeout_thresh = 20, .rxfifo_full_thresh = 50, }; ESP_ERROR_CHECK(uart_intr_config(uart_port, &intr_config)); uart_enable_intr_mask(uart_port, uart_intr); uart_set_sw_flow_ctrl(uart_port, true, 10, 30); uart_enable_pattern_det_baud_intr(uart_port, '\n', 1, 10, 0, 0); xTaskCreate(uart_intr_callback, "uart_intr", 4096, NULL, 5, NULL); xTaskCreate(stream_buffer_info, "stream_buffer_info", 4096, NULL, 10, NULL); ESP_LOGI(TAG, "UART初始化成功，波特率为：%d", uart_config.baud_rate); } void uart_intr_callback(void *arg) { uart_event_t uart_event; size_t wait_read_size; int len; char read_data[1024]; while(1) { xQueueReceive(uart_intr_handle, &uart_event, portMAX_DELAY); switch (uart_event.type) { case UART_DATA: if (uart_event.timeout_flag == false){ ESP_LOGI(TAG, "触发接收FIFO阈值中断, uart rx data size: %d", uart_event.size); } else{ ESP_LOGI(TAG, "触发接收超时中断, uart rx data size: %d", uart_event.size); } break; case UART_PATTERN_DET: ESP_LOGI(TAG, "触发PATTERN_DET中断, uart rx data size: %d", uart_event.size); break; default: ESP_LOGI(TAG, "其他未知事件触发"); break; } uart_get_buffered_data_len(uart_port, &wait_read_size); len = uart_read_bytes(uart_port, read_data, wait_read_size, 1000); if(len > 0){ read_data[len] = '\0'; ESP_LOGI(TAG, "uart rx data: %s", read_data); } xStreamBufferSend(uart_stream_buffer, read_data, len, 0); } } void stream_buffer_info(void *arg) { char data[1024]; while(1) { ESP_LOGI(TAG, "Stream Buffer中包含的数据大小: %d", xStreamBufferBytesAvailable(uart_stream_buffer)); size_t rx_data_size = xStreamBufferReceive(uart_stream_buffer, data, 30, portMAX_DELAY); ESP_LOGI(TAG, "Stream Buffer中收到的数据大小: %d", rx_data_size); rx_data_size = uart_write_bytes(uart_port, data, rx_data_size); ESP_LOGI(TAG, "uart tx length: %d", rx_data_size); } }

2.2 uart_m.h

#ifndef UART_M_H #define UART_M_H #include <string.h> // 字符串处理函数 #include "esp_log.h" // ESP32日志函数 #include "FreeRTOS/FreeRTOS.h" // FreeRTOS函数 #include "FreeRTOS/task.h" // FreeRTOS任务管理函数 #include "FreeRTOS/semphr.h" // FreeRTOS信号量管理函数 #include "hal/uart_ll.h" #define uart_rx_buffer 512 #define uart_tx_buffer 512 #define uart_port UART_NUM_0 #define uart_tx_pin GPIO_NUM_37 #define uart_rx_pin GPIO_NUM_38 #define uart_intr UART_INTR_RXFIFO_TOUT | UART_INTR_RXFIFO_FULL void uart_init(void); #endif // UART_M_H

2.3 main.c

#include <stdio.h> #include "user.h" #include "uart_m.h" void app_main(void) { CONSOLE_REPL_INIT(); // 初始化控制台REPL环境 uart_init(); // 初始化UART while(1) { vTaskDelay(pdMS_TO_TICKS(1000)); } }

2.4 代码解释

这里面关于uart的API就不解释了，需要了解的话去看第三讲的工程。这里说一下改动的地方。

在初始化函数uart_init里使用函数xStreamBufferCreate创建一个流缓冲区，这里没使用带回调函数，一帮情况下也不需要用回调。创建的缓冲区大小是1024字节，触发阈值是100字节。接着在尾部在创建一个流缓冲区任务stream_buffer_info。

在原先串口任务中调用xStreamBufferSend函数，将接收到的数据发送到流缓冲区。

在任务stream_buffer_info里，先使用xStreamBufferSpacesAvailable查询流缓冲区中空闲的大小来判断是否会溢出，接着使用xStreamBufferReceive来接收数据。如果流缓冲区中的数据大小没有到阈值，就会阻塞等待。

xStreamBufferReceive函数读取机制是，第一次读取时缓冲区内数据没有达到阈值，则会阻塞等待，直到数据达到阈值，接触阻塞并读取设定的字节数，如果接下来再次调用则不会阻塞，直到将流缓冲区所有数据读完，下次读取时继续阻塞，如此周期运行。

所以流缓冲区一旦满足了100个字节的阈值，立马解除阻塞并从缓冲区读取30个字节的数据并回传到上位机。接着不断读取30个（不够30个就读所有剩余），直到流缓冲区数据为0，从新下一次阻塞。

2.5 结果展示

我每次从上位机发送21个字节到设备，第五次发送完流缓冲区数据大小为105字节，大于阈值100字节，解除任务stream_buffer_info阻塞，接着不断循环，每次30个字节从流缓冲区中读出，最后一次剩余15个字节，不满30，则全部读出。此时流缓冲区内为空，则继续阻塞任务stream_buffer_info，等待下一次阈值到来。

三、环形缓冲区 (Ring Buffer)使用

敬请期待