14.FreeRTOS 流媒体缓存 Stream Buffer

FreeRTOS 中的 Stream Buffer（流媒体缓存）

在实时操作系统（RTOS）中，处理流媒体数据是一项非常关键的任务。FreeRTOS 提供了一种名为 Stream Buffer（流媒体缓存）的机制，用于高效地管理和传输流式数据，如音频、视频等。

什么是 Stream Buffer？

Stream Buffer 是 FreeRTOS 提供的一种数据结构，专门用于在任务之间传递和处理流式数据。与队列不同，Stream Buffer 适用于大数据块的流式传输，而不是单个数据项的传输。这使得它非常适合用于音频、视频和其他连续数据流的场景。

Stream Buffer 的工作原理

Stream Buffer 的基本操作包括创建、发送和接收数据。它通过一个环形缓冲区来存储数据，使得数据可以连续地写入和读取。以下是其主要操作：

1. 创建 Stream Buffer：使用 xStreamBufferCreate 函数创建一个新的 Stream Buffer。
2. 发送数据到 Stream Buffer：任务使用 xStreamBufferSend 将数据发送到缓冲区。
3. 从 Stream Buffer 接收数据：任务使用 xStreamBufferReceive 从缓冲区接收数据。

流媒体传输任务处理的示例

下面是一个示例，展示了如何使用 Stream Buffer 实现流水线式任务处理：

#include <Arduino.h>
#include <FreeRTOS.h>
#include <stream_buffer.h>#define BUFFER_SIZE 1024
#define TRIGGER_LEVEL 1StreamBufferHandle_t xStreamBuffer;// Task1: 发送数据到 Stream Buffer
void Task1(void *pvParameters) {const char *data = "Task 1 data";for (;;) {// 模拟 Task1 的工作delay(1000);Serial.println("Task 1 is sending data");xStreamBufferSend(xStreamBuffer, data, strlen(data), portMAX_DELAY);}
}// Task2: 从 Stream Buffer 接收数据并处理
void Task2(void *pvParameters) {char buffer[50];for (;;) {// 从 Stream Buffer 接收数据size_t bytesReceived = xStreamBufferReceive(xStreamBuffer, buffer, sizeof(buffer), portMAX_DELAY);if (bytesReceived > 0) {buffer[bytesReceived] = '\0'; // 确保字符串以 NULL 结尾Serial.print("Task 2 received data: ");Serial.println(buffer);// 模拟 Task2 的工作delay(1000);}}
}void setup() {Serial.begin(115200);// 创建 Stream BufferxStreamBuffer = xStreamBufferCreate(BUFFER_SIZE, TRIGGER_LEVEL);if (xStreamBuffer == NULL) {// 创建失败，处理错误while (1);}// 创建任务xTaskCreate(Task1, "Task 1", 10000, NULL, 1, NULL);xTaskCreate(Task2, "Task 2", 10000, NULL, 1, NULL);// 启动任务调度器vTaskStartScheduler();
}void loop() {// loop 函数为空，不需要额外的代码
}

API

使用 Stream Buffer 的步骤

1. 创建 Stream Buffer

要创建一个 Stream Buffer，使用 xStreamBufferCreate 函数：

StreamBufferHandle_t xStreamBufferCreate(size_t xBufferSizeBytes, size_t xTriggerLevelBytes);

参数：

• xBufferSizeBytes：缓冲区的大小（字节数）。
• xTriggerLevelBytes：触发接收任务的最小字节数。

返回值：

• 成功时返回 Stream Buffer 的句柄（非 NULL）；失败时返回 NULL。

示例：

StreamBufferHandle_t xStreamBuffer = xStreamBufferCreate(1024, 1);
if (xStreamBuffer == NULL) {// 创建失败，处理错误
}

2. 发送数据到 Stream Buffer

使用 xStreamBufferSend 将数据发送到缓冲区：

size_t xStreamBufferSend(StreamBufferHandle_t xStreamBuffer, const void *pvTxData, size_t xDataLengthBytes, TickType_t xTicksToWait);

参数：

• xStreamBuffer：Stream Buffer 的句柄。
• pvTxData：要发送的数据指针。
• xDataLengthBytes：发送的数据长度（字节数）。
• xTicksToWait：等待时间（滴答数）。

返回值：

• 成功发送的数据字节数。如果返回值小于 xDataLengthBytes，则表示在 xTicksToWait 时间内没有足够的空间发送所有数据。

示例：

const char *data = "Hello, FreeRTOS!";
size_t bytesSent = xStreamBufferSend(xStreamBuffer, data, strlen(data), portMAX_DELAY);
if (bytesSent < strlen(data)) {// 发送的数据不完整，处理错误
}

3. 从 Stream Buffer 接收数据

使用 xStreamBufferReceive 从缓冲区接收数据：

size_t xStreamBufferReceive(StreamBufferHandle_t xStreamBuffer, void *pvRxData, size_t xBufferLengthBytes, TickType_t xTicksToWait);

参数：

• xStreamBuffer：Stream Buffer 的句柄。
• pvRxData：接收数据的缓冲区指针。
• xBufferLengthBytes：接收缓冲区的大小（字节数）。
• xTicksToWait：等待时间（滴答数）。

返回值：

• 成功接收的数据字节数。如果返回值小于 xBufferLengthBytes，则表示在 xTicksToWait 时间内没有接收到足够的数据。

示例：

char buffer[50];
size_t bytesReceived = xStreamBufferReceive(xStreamBuffer, buffer, sizeof(buffer), portMAX_DELAY);
if (bytesReceived > 0) {buffer[bytesReceived] = '\0'; // 确保字符串以 NULL 结尾Serial.println(buffer);
} else {// 接收失败，处理错误
}

其他 Stream Buffer API

1. xStreamBufferCreateStatic

创建一个静态分配的 Stream Buffer。

StreamBufferHandle_t xStreamBufferCreateStatic(size_t xBufferSizeBytes, size_t xTriggerLevelBytes, uint8_t *pucStreamBufferStorageArea, StaticStreamBuffer_t *pxStaticStreamBuffer);

参数：

• xBufferSizeBytes：缓冲区的大小（字节数）。
• xTriggerLevelBytes：触发接收任务的最小字节数。
• pucStreamBufferStorageArea：指向预先分配的缓冲区存储区。
• pxStaticStreamBuffer：指向预先分配的 StaticStreamBuffer_t 结构。

返回值：

• 成功时返回 Stream Buffer 的句柄（非 NULL）；失败时返回 NULL。

2. xStreamBufferReset

重置 Stream Buffer，使其变为空。

BaseType_t xStreamBufferReset(StreamBufferHandle_t xStreamBuffer);

参数：

• xStreamBuffer：Stream Buffer 的句柄。

返回值：

• 如果成功重置 Stream Buffer，则返回 pdPASS；如果在 Stream Buffer 上有任务阻塞，则返回 pdFAIL。

示例：

if (xStreamBufferReset(xStreamBuffer) == pdPASS) {// 重置成功
} else {// 重置失败，处理错误
}

3. xStreamBufferSpacesAvailable

获取 Stream Buffer 中可用的空闲空间（字节数）。

size_t xStreamBufferSpacesAvailable(StreamBufferHandle_t xStreamBuffer);

参数：

• xStreamBuffer：Stream Buffer 的句柄。

返回值：

• Stream Buffer 中当前可用的空闲空间，以字节为单位。

示例：

size_t spaceAvailable = xStreamBufferSpacesAvailable(xStreamBuffer);
Serial.print("Space available: ");
Serial.println(spaceAvailable);

4. xStreamBufferBytesAvailable

获取 Stream Buffer 中可读取的字节数。

size_t xStreamBufferBytesAvailable(StreamBufferHandle_t xStreamBuffer);

参数：

• xStreamBuffer：Stream Buffer 的句柄。

返回值：

• Stream Buffer 中当前可读取的字节数。

示例：

size_t bytesAvailable = xStreamBufferBytesAvailable(xStreamBuffer);
Serial.print("Bytes available: ");
Serial.println(bytesAvailable);

5. xStreamBufferSetTriggerLevel

设置触发接收任务的最小字节数。

BaseType_t xStreamBufferSetTriggerLevel(StreamBufferHandle_t xStreamBuffer, size_t xTriggerLevel);

参数：

• xStreamBuffer：Stream Buffer 的句柄。
• xTriggerLevel：触发级别（字节数）。

返回值：

• 如果成功设置触发级别，则返回 pdPASS；如果提供的 xTriggerLevel 超过缓冲区大小，则返回 pdFAIL。

示例：

if (xStreamBufferSetTriggerLevel(xStreamBuffer, 10) == pdPASS) {// 设置成功
} else {// 设置失败，处理错误
}

6. vStreamBufferDelete

删除 Stream Buffer，释放分配的内存。

void vStreamBufferDelete(StreamBufferHandle_t xStreamBuffer);

参数：

• xStreamBuffer：Stream Buffer 的句柄。

返回值：

• 无返回值。

示例：

vStreamBufferDelete(xStreamBuffer);

结论

FreeRTOS 的 Stream Buffer 提供了一种高效的机制来处理和传输流媒体数据。通过合理使用 Stream Buffer，可以在嵌入式系统中实现稳定可靠的流媒体数据处理。在实际应用中，我们应根据具体需求调整缓冲区大小和触发级别，并优化数据处理流程，以实现最佳性能。