基于STM32的边缘计算实时数据处理可视化系统:嵌入式C++、 FreeRTOS、Kafka、Spring Boo、 InfluxDB、Grafana（代码示例）

一、项目概述

本项目旨在设计并实现一个基于STM32的边缘计算实时数据处理系统。该系统能够在边缘设备端进行数据采集、预处理,并将处理后的数据实时传输到后端服务器进行进一步分析和存储。

本项目主要解决以下问题:

1. 减轻后端服务器的数据处理负担,提高系统整体效率

2. 降低数据传输带宽需求,减少通信成本

3. 实现近实时的数据分析,提高系统响应速度

4. 增强数据隐私保护,敏感数据可在本地处理后再传输

通过边缘计算与实时数据处理相结合,本系统可广泛应用于工业物联网、智能家居、环境监测等多个领域,为用户提供高效、安全、实时的数据分析服务。

二、系统架构

本系统采用分层架构设计,主要包括边缘层、网络层和云端层三个部分。

2.1 硬件选型

• 边缘设备: STM32F407VGT6 微控制器

• 传感器: DHT22温湿度传感器、BMP280气压传感器

• 通信模块: ESP8266 Wi-Fi模块

2.2 软件技术栈

• 边缘端: FreeRTOS实时操作系统、STM32 HAL库

• 网络传输: MQTT协议

• 后端服务: Apache Kafka消息队列、Spring Boot应用服务器

• 数据存储: InfluxDB时序数据库

• 数据可视化: Grafana仪表板

2.3 系统架构图

三、环境搭建

3.1 开发环境

• STM32CubeIDE 1.8.0

• STM32CubeMX 6.5.0

• FreeRTOS 10.3.1

• ESP8266 AT固件 v3.0.0

3.2 STM32环境配置

1. 安装STM32CubeIDE

2. 使用STM32CubeMX生成项目框架

3. 配置时钟、GPIO、UART等外设

4. 启用FreeRTOS支持

配置示例:

// 时钟配置
RCC_OscInitTypeDef RCC_OscInitStruct = {0};
RCC_ClkInitTypeDef RCC_ClkInitStruct = {0};

RCC_OscInitStruct.OscillatorType = RCC_OSCILLATORTYPE_HSE;
RCC_OscInitStruct.HSEState = RCC_HSE_ON;
RCC_OscInitStruct.PLL.PLLState = RCC_PLL_ON;
RCC_OscInitStruct.PLL.PLLSource = RCC_PLLSOURCE_HSE;
RCC_OscInitStruct.PLL.PLLM = 8;
RCC_OscInitStruct.PLL.PLLN = 336;
RCC_OscInitStruct.PLL.PLLP = RCC_PLLP_DIV2;
RCC_OscInitStruct.PLL.PLLQ = 7;

RCC_ClkInitStruct.ClockType = RCC_CLOCKTYPE_HCLK|RCC_CLOCKTYPE_SYSCLK
                              |RCC_CLOCKTYPE_PCLK1|RCC_CLOCKTYPE_PCLK2;
RCC_ClkInitStruct.SYSCLKSource = RCC_SYSCLKSOURCE_PLLCLK;
RCC_ClkInitStruct.AHBCLKDivider = RCC_SYSCLK_DIV1;
RCC_ClkInitStruct.APB1CLKDivider = RCC_HCLK_DIV4;
RCC_ClkInitStruct.APB2CLKDivider = RCC_HCLK_DIV2;

HAL_RCC_OscConfig(&RCC_OscInitStruct);
HAL_RCC_ClockConfig(&RCC_ClkInitStruct, FLASH_LATENCY_5);

// UART配置
huart2.Instance = USART2;
huart2.Init.BaudRate = 115200;
huart2.Init.WordLength = UART_WORDLENGTH_8B;
huart2.Init.StopBits = UART_STOPBITS_1;
huart2.Init.Parity = UART_PARITY_NONE;
huart2.Init.Mode = UART_MODE_TX_RX;
huart2.Init.HwFlowCtl = UART_HWCONTROL_NONE;
huart2.Init.OverSampling = UART_OVERSAMPLING_16;
HAL_UART_Init(&huart2);

// FreeRTOS配置
osThreadDef(defaultTask, StartDefaultTask, osPriorityNormal, 0, 128);
defaultTaskHandle = osThreadCreate(osThread(defaultTask), NULL);

3.3 后端环境配置

1. 安装Java JDK 11+

2. 安装Apache Kafka 2.8.0

3. 安装InfluxDB 2.0

4. 安装Grafana 8.0

Kafka配置示例 (server.properties):

broker.id=0
listeners=PLAINTEXT://localhost:9092
num.network.threads=3
num.io.threads=8
socket.send.buffer.bytes=102400
socket.receive.buffer.bytes=102400
socket.request.max.bytes=104857600
log.dirs=/tmp/kafka-logs
num.partitions=1
num.recovery.threads.per.data.dir=1
offsets.topic.replication.factor=1
transaction.state.log.replication.factor=1
transaction.state.log.min.isr=1
log.retention.hours=168
log.segment.bytes=1073741824
log.retention.check.interval.ms=300000

注意事项:

• 确保所有服务的端口不冲突

• 配置适当的安全策略,如设置强密码和启用SSL加密

• 根据系统负载调整Kafka和InfluxDB的性能参数

四、代码实现

4.1 STM32边缘设备数据采集与预处理

#include "main.h"
#include "cmsis_os.h"
#include "dht22.h"
#include "bmp280.h"

// 定义传感器数据结构
typedef struct {
    float temperature;
    float humidity;
    float pressure;
} SensorData;

// 数据采集任务
void DataCollectionTask(void const * argument)
{
    SensorData data;
    for(;;)
    {
        // 读取DHT22温湿度数据
        DHT22_Read(&data.temperature, &data.humidity);
        
        // 读取BMP280气压数据
        data.pressure = BMP280_ReadPressure();
        
        // 数据预处理 (示例: 简单的移动平均滤波)
        static float temp_buffer[5] = {0};
        static int buffer_index = 0;
        
        temp_buffer[buffer_index] = data.temperature;
        buffer_index = (buffer_index + 1) % 5;
        
        float avg_temp = 0;
        for(int i = 0; i < 5; i++) {
            avg_temp += temp_buffer[i];
        }
        data.temperature = avg_temp / 5;
        
        // 发送数据到MQTT发布任务
        xQueueSend(mqttPublishQueue, &data, portMAX_DELAY);
        
        osDelay(5000); // 每5秒采集一次数据
    }
}

// MQTT发布任务
void MQTTPublishTask(void const * argument)
{
    SensorData data;
    char mqtt_message[100];
    for(;;)
    {
        if(xQueueReceive(mqttPublishQueue, &data, portMAX_DELAY) == pdTRUE)
        {
            // 格式化MQTT消息
            snprintf(mqtt_message, sizeof(mqtt_message), 
                     "{\"temp\":%.2f,\"hum\":%.2f,\"press\":%.2f}",
                     data.temperature, data.humidity, data.pressure);
            
            // 通过ESP8266发送MQTT消息
            ESP8266_MQTT_Publish("sensor/data", mqtt_message);
        }
    }
}

4.2 Spring Boot后端服务

@Service
public class SensorDataService {

    private final KafkaTemplate<String, String> kafkaTemplate;
    private final InfluxDBClient influxDBClient;

    @Autowired
    public SensorDataService(KafkaTemplate<String, String> kafkaTemplate, InfluxDBClient influxDBClient) {
        this.kafkaTemplate = kafkaTemplate;
        this.influxDBClient = influxDBClient;
    }

    @KafkaListener(topics = "sensor.data", groupId = "sensor-group")
    public void consumeSensorData(String message) {
        try {
            JSONObject jsonObject = new JSONObject(message);
            double temperature = jsonObject.getDouble("temp");
            double humidity = jsonObject.getDouble("hum");
            double pressure = jsonObject.getDouble("press");

            // 存储数据到InfluxDB
            Point point = Point.measurement("sensor_data")
                    .addTag("sensor_id", "stm32_01")
                    .addField("temperature", temperature)
                    .addField("humidity", humidity)
                    .addField("pressure", pressure)
                    .time(Instant.now(), WritePrecision.NS);

            WriteApiBlocking writeApi = influxDBClient.getWriteApiBlocking();
            writeApi.writePoint("sensor_bucket", "org", point);

            // 进行数据分析 (示例: 简单的阈值检测)
            if (temperature > 30 || humidity > 80) {
                sendAlert("High temperature or humidity detected!");
            }

        } catch (JSONException e) {
            log.error("Error parsing sensor data: ", e);
        }
    }

    private void sendAlert(String message) {
        // 实现告警逻辑,如发送邮件或推送通知
        log.warn("Alert: " + message);
    }
}

4.3 代码说明

STM32边缘设备代码

1. 数据结构:

   • SensorData结构体用于存储温度、湿度和气压数据。

2. 数据采集任务(DataCollectionTask):

   • 定期从DHT22和BMP280传感器读取数据。

   • 对温度数据进行简单的移动平均滤波，作为数据预处理的示例。

   • 使用FreeRTOS队列(xQueueSend)将处理后的数据发送到MQTT发布任务。

3. MQTT发布任务(MQTTPublishTask):

   • 从队列接收传感器数据。

   • 将数据格式化为JSON字符串。

   • 使用ESP8266模块通过MQTT协议发布数据。

Spring Boot后端服务代码

1. 依赖注入:

   • 使用@Autowired注入KafkaTemplate和InfluxDBClient。

2. Kafka消费者:

   • @KafkaListener注解用于订阅"sensor.data"主题。

   • consumeSensorData方法处理接收到的传感器数据。

3. 数据处理:

   • 解析JSON格式的传感器数据。

   • 创建InfluxDB的Point对象，添加测量值、标签和字段。

4. 数据存储:

   • 使用InfluxDB的WriteApiBlocking将数据点写入数据库。

5. 简单分析:

   • 实现了基本的阈值检测，当温度或湿度超过设定值时触发告警。

6. 错误处理:

   • 使用try-catch块捕获JSON解析错误，并记录日志。

这些代码实现了从STM32边缘设备采集数据，通过MQTT传输，再由Spring Boot后端接收并存储到InfluxDB的完整流程。系统设计考虑了实时性、可靠性和可扩展性，为物联网应用提供了一个solid的基础架构。

4.4 Grafana数据可视化

1. 在Grafana中添加InfluxDB数据源

2. 创建新的Dashboard

3. 添加Panel,使用InfluxQL查询语言配置数据展示

示例查询:

SELECT mean("temperature") AS "avg_temp", 
       mean("humidity") AS "avg_hum", 
       mean("pressure") AS "avg_press"
FROM "sensor_data"
WHERE $timeFilter
GROUP BY time($__interval)

五、项目总结

本项目成功实现了一个基于STM32的边缘计算实时数据处理系统,主要功能和实现过程如下:

1. 数据采集: 使用DHT22和BMP280传感器采集温度、湿度和气压数据。

2. 边缘处理: 在STM32上进行数据预处理,如滤波和压缩。

3. 数据传输: 利用ESP8266 Wi-Fi模块和MQTT协议将数据实时传输到云端。

4. 消息队列: 使用Apache Kafka实现高吞吐量的消息处理。

5. 后端服务: Spring Boot应用接收并处理传感器数据,实现数据存储和简单的分析。

6. 数据存储: 采用InfluxDB时序数据库存储大量传感器数据。

7. 数据可视化: 使用Grafana创建直观的数据仪表板。

通过这个系统,我们成功地将数据处理前移到边缘设备,减轻了后端服务器的负担,同时实现了近实时的数据分析和可视化。该系统具有良好的可扩展性,可以根据需求添加更多的传感器和分析功能。