基于STM32的边缘计算实时数据处理可视化系统:嵌入式C++、 FreeRTOS、Kafka ...

王國慶  金牌会员 | 2024-12-8 21:24:20 | 显示全部楼层 | 阅读模式
打印 上一主题 下一主题
楼主

主题 915|帖子 915|积分 2749

一、项目概述

本项目旨在设计并实现一个基于STM32的边缘计算实时数据处理系统。该系统可以或许在边缘装备端举行数据采集、预处理,并将处理后的数据实时传输到后端服务器举行进一步分析和存储。
本项目主要解决以下问题:

  • 减轻后端服务器的数据处理负担,提高系统整体效率
  • 降低数据传输带宽需求,淘汰通讯成本
  • 实现近实时的数据分析,提高系统相应速度
  • 加强数据隐私掩护,敏感数据可在当地处理后再传输
通过边缘计算与实时数据处理相结合,本系统可广泛应用于工业物联网、智能家居、环境监测等多个范畴,为用户提供高效、安全、实时的数据分析服务。
二、系统架构

本系统采用分层架构设计,主要包括边缘层、网络层和云端层三个部门。
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环境配置


  • 安装STM32CubeIDE
  • 使用STM32CubeMX天生项目框架
  • 配置时钟、GPIO、UART等外设
  • 启用FreeRTOS支持
配置示例:
  1. // 时钟配置
  2. RCC_OscInitTypeDef RCC_OscInitStruct = {0};
  3. RCC_ClkInitTypeDef RCC_ClkInitStruct = {0};
  5. RCC_OscInitStruct.OscillatorType = RCC_OSCILLATORTYPE_HSE;
  6. RCC_OscInitStruct.HSEState = RCC_HSE_ON;
  7. RCC_OscInitStruct.PLL.PLLState = RCC_PLL_ON;
  8. RCC_OscInitStruct.PLL.PLLSource = RCC_PLLSOURCE_HSE;
  9. RCC_OscInitStruct.PLL.PLLM = 8;
  10. RCC_OscInitStruct.PLL.PLLN = 336;
  11. RCC_OscInitStruct.PLL.PLLP = RCC_PLLP_DIV2;
  12. RCC_OscInitStruct.PLL.PLLQ = 7;
  14. RCC_ClkInitStruct.ClockType = RCC_CLOCKTYPE_HCLK|RCC_CLOCKTYPE_SYSCLK
  15.                               |RCC_CLOCKTYPE_PCLK1|RCC_CLOCKTYPE_PCLK2;
  16. RCC_ClkInitStruct.SYSCLKSource = RCC_SYSCLKSOURCE_PLLCLK;
  17. RCC_ClkInitStruct.AHBCLKDivider = RCC_SYSCLK_DIV1;
  18. RCC_ClkInitStruct.APB1CLKDivider = RCC_HCLK_DIV4;
  19. RCC_ClkInitStruct.APB2CLKDivider = RCC_HCLK_DIV2;
  21. HAL_RCC_OscConfig(&RCC_OscInitStruct);
  22. HAL_RCC_ClockConfig(&RCC_ClkInitStruct, FLASH_LATENCY_5);
  24. // UART配置
  25. huart2.Instance = USART2;
  26. huart2.Init.BaudRate = 115200;
  27. huart2.Init.WordLength = UART_WORDLENGTH_8B;
  28. huart2.Init.StopBits = UART_STOPBITS_1;
  29. huart2.Init.Parity = UART_PARITY_NONE;
  30. huart2.Init.Mode = UART_MODE_TX_RX;
  31. huart2.Init.HwFlowCtl = UART_HWCONTROL_NONE;
  32. huart2.Init.OverSampling = UART_OVERSAMPLING_16;
  33. HAL_UART_Init(&huart2);
  35. // FreeRTOS配置
  36. osThreadDef(defaultTask, StartDefaultTask, osPriorityNormal, 0, 128);
  37. defaultTaskHandle = osThreadCreate(osThread(defaultTask), NULL);
复制代码
3.3 后端环境配置


  • 安装Java JDK 11+
  • 安装Apache Kafka 2.8.0
  • 安装InfluxDB 2.0
  • 安装Grafana 8.0
Kafka配置示例 (server.properties):
  1. broker.id=0
  2. listeners=PLAINTEXT://localhost:9092
  3. num.network.threads=3
  4. num.io.threads=8
  5. socket.send.buffer.bytes=102400
  6. socket.receive.buffer.bytes=102400
  7. socket.request.max.bytes=104857600
  8. log.dirs=/tmp/kafka-logs
  9. num.partitions=1
  10. num.recovery.threads.per.data.dir=1
  11. offsets.topic.replication.factor=1
  12. transaction.state.log.replication.factor=1
  13. transaction.state.log.min.isr=1
  14. log.retention.hours=168
  15. log.segment.bytes=1073741824
  16. log.retention.check.interval.ms=300000
复制代码
注意事项:


  • 确保所有服务的端口不辩论
  • 配置适当的安全策略,如设置强暗码和启用SSL加密
  • 根据系统负载调整Kafka和InfluxDB的性能参数
四、代码实现

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

  1. #include "main.h"
  2. #include "cmsis_os.h"
  3. #include "dht22.h"
  4. #include "bmp280.h"
  6. // 定义传感器数据结构
  7. typedef struct {
  8.     float temperature;
  9.     float humidity;
  10.     float pressure;
  11. } SensorData;
  13. // 数据采集任务
  14. void DataCollectionTask(void const * argument)
  15. {
  16.     SensorData data;
  17.     for(;;)
  18.     {
  19.         // 读取DHT22温湿度数据
  20.         DHT22_Read(&data.temperature, &data.humidity);
  21.         
  22.         // 读取BMP280气压数据
  23.         data.pressure = BMP280_ReadPressure();
  24.         
  25.         // 数据预处理 (示例: 简单的移动平均滤波)
  26.         static float temp_buffer[5] = {0};
  27.         static int buffer_index = 0;
  28.         
  29.         temp_buffer[buffer_index] = data.temperature;
  30.         buffer_index = (buffer_index + 1) % 5;
  31.         
  32.         float avg_temp = 0;
  33.         for(int i = 0; i < 5; i++) {
  34.             avg_temp += temp_buffer[i];
  35.         }
  36.         data.temperature = avg_temp / 5;
  37.         
  38.         // 发送数据到MQTT发布任务
  39.         xQueueSend(mqttPublishQueue, &data, portMAX_DELAY);
  40.         
  41.         osDelay(5000); // 每5秒采集一次数据
  42.     }
  43. }
  45. // MQTT发布任务
  46. void MQTTPublishTask(void const * argument)
  47. {
  48.     SensorData data;
  49.     char mqtt_message[100];
  50.     for(;;)
  51.     {
  52.         if(xQueueReceive(mqttPublishQueue, &data, portMAX_DELAY) == pdTRUE)
  53.         {
  54.             // 格式化MQTT消息
  55.             snprintf(mqtt_message, sizeof(mqtt_message),
  56.                      "{"temp":%.2f,"hum":%.2f,"press":%.2f}",
  57.                      data.temperature, data.humidity, data.pressure);
  58.             
  59.             // 通过ESP8266发送MQTT消息
  60.             ESP8266_MQTT_Publish("sensor/data", mqtt_message);
  61.         }
  62.     }
  63. }
复制代码
4.2 Spring Boot后端服务

  1. @Service
  2. public class SensorDataService {
  4.     private final KafkaTemplate<String, String> kafkaTemplate;
  5.     private final InfluxDBClient influxDBClient;
  7.     @Autowired
  8.     public SensorDataService(KafkaTemplate<String, String> kafkaTemplate, InfluxDBClient influxDBClient) {
  9.         this.kafkaTemplate = kafkaTemplate;
  10.         this.influxDBClient = influxDBClient;
  11.     }
  13.     @KafkaListener(topics = "sensor.data", groupId = "sensor-group")
  14.     public void consumeSensorData(String message) {
  15.         try {
  16.             JSONObject jsonObject = new JSONObject(message);
  17.             double temperature = jsonObject.getDouble("temp");
  18.             double humidity = jsonObject.getDouble("hum");
  19.             double pressure = jsonObject.getDouble("press");
  21.             // 存储数据到InfluxDB
  22.             Point point = Point.measurement("sensor_data")
  23.                     .addTag("sensor_id", "stm32_01")
  24.                     .addField("temperature", temperature)
  25.                     .addField("humidity", humidity)
  26.                     .addField("pressure", pressure)
  27.                     .time(Instant.now(), WritePrecision.NS);
  29.             WriteApiBlocking writeApi = influxDBClient.getWriteApiBlocking();
  30.             writeApi.writePoint("sensor_bucket", "org", point);
  32.             // 进行数据分析 (示例: 简单的阈值检测)
  33.             if (temperature > 30 || humidity > 80) {
  34.                 sendAlert("High temperature or humidity detected!");
  35.             }
  37.         } catch (JSONException e) {
  38.             log.error("Error parsing sensor data: ", e);
  39.         }
  40.     }
  42.     private void sendAlert(String message) {
  43.         // 实现告警逻辑,如发送邮件或推送通知
  44.         log.warn("Alert: " + message);
  45.     }
  46. }
复制代码
4.3 代码阐明

STM32边缘装备代码


  • 数据结构:

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

  • 数据采集任务(DataCollectionTask):

    • 定期从DHT22和BMP280传感器读取数据。
    • 对温度数据举行简朴的移动平均滤波，作为数据预处理的示例。
    • 使用FreeRTOS队列(xQueueSend)将处理后的数据发送到MQTT发布任务。

  • MQTT发布任务(MQTTPublishTask):

    • 从队列吸收传感器数据。
    • 将数据格式化为JSON字符串。
    • 使用ESP8266模块通过MQTT协议发布数据。

Spring Boot后端服务代码


  • 依赖注入:

    • 使用@Autowired注入KafkaTemplate和InfluxDBClient。

  • Kafka斲丧者:

    • @KafkaListener注解用于订阅"sensor.data"主题。
    • consumeSensorData方法处理吸收到的传感器数据。

  • 数据处理:

    • 解析JSON格式的传感器数据。
    • 创建InfluxDB的Point对象，添加丈量值、标签和字段。

  • 数据存储:

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

  • 简朴分析:

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

  • 错误处理:

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

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


  • 在Grafana中添加InfluxDB数据源
  • 创建新的Dashboard
  • 添加Panel,使用InfluxQL查询语言配置数据展示
示例查询:
  1. SELECT mean("temperature") AS "avg_temp",
  2.        mean("humidity") AS "avg_hum",
  3.        mean("pressure") AS "avg_press"
  4. FROM "sensor_data"
  5. WHERE $timeFilter
  6. GROUP BY time($__interval)
复制代码
五、项目总结

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

  • 数据采集: 使用DHT22和BMP280传感器采集温度、湿度和睦压数据。
  • 边缘处理: 在STM32上举行数据预处理,如滤波和压缩。
  • 数据传输: 利用ESP8266 Wi-Fi模块和MQTT协议将数据实时传输到云端。
  • 消息队列: 使用Apache Kafka实现高吞吐量的消息处理。
  • 后端服务: Spring Boot应用吸收并处理传感器数据,实现数据存储和简朴的分析。
  • 数据存储: 采用InfluxDB时序数据库存储大量传感器数据。
  • 数据可视化: 使用Grafana创建直观的数据仪表板。
通过这个系统,我们成功地将数据处理前移到边缘装备,减轻了后端服务器的负担,同时实现了近实时的数据分析和可视化。该系统具有精良的可扩展性,可以根据需求添加更多的传感器和分析功能。

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。
回复

使用道具 举报

0 个回复

倒序浏览
返回列表

快速回复

高级模式
B Color Image Link Quote Code Smilies
您需要登录后才可以回帖 登录 or 立即注册

本版积分规则

王國慶

金牌会员
这个人很懒什么都没写!

楼主热帖

标签云

挺好的 服务器
微信订阅号
微信服务号
微信客服
小程序
H5
关于我们 商务合作 网站地图
©Copyright 2022-2024 杭州祥升科技有限公司 版权所有 浙ICP备20004199号
快速回复 返回顶部 返回列表