如何向MQTT Broker批量发送ESP32读取的MPU6050传感器数据

要实现每10秒批量发送100个MPU6050的AcY数值，你可以通过数据缓存+定时批量发送的方式实现，核心思路是先将传感器读取到的数据存入列表，当满足数量或时间条件时，一次性将所有数据打包发送到MQTT Broker。

修改后的代码 #

from umqttsimple import MQTTClient
from machine import SoftI2C, Pin
import mpu6050
import time
import json  # 用于将列表转为JSON格式，方便Node-RED解析

# 初始化I2C和MPU6050
i2c = SoftI2C(scl=Pin(21), sda=Pin(22))
mpu = mpu6050.accel(i2c)

# MQTT配置（请替换为你的实际信息）
mqtt_server = "你的MQTT服务器地址"
mqtt_username = "你的用户名"
mqtt_password = "你的密码"
topic = b"你的MQTT主题"

client = MQTTClient(b"bruecke", mqtt_server, 1883, user=mqtt_username, password=mqtt_password)
client.connect()

data_buffer = []  # 缓存传感器数据的列表
start_time = time.time()  # 记录批次开始时间
batch_size = 100  # 单次批量发送的数据量
interval = 10  # 批量发送的时间间隔（秒）

while True:
    # 读取传感器AcY数值
    werte = mpu.get_values()
    acy_value = werte['AcY']
    data_buffer.append(acy_value)
    print(acy_value)
    
    # 检查是否满足发送条件：缓存满100条 或 时间超过10秒
    current_time = time.time()
    if len(data_buffer) >= batch_size or (current_time - start_time) >= interval:
        # 将数据列表转为JSON字符串（Node-RED可直接解析）
        payload = json.dumps(data_buffer)
        # 批量发布数据
        client.publish(topic, payload)
        print(f"已批量发送{len(data_buffer)}个数据")
        # 重置缓存和计时
        data_buffer = []
        start_time = current_time
    
    # 控制读取频率，确保10秒内可收集约100条数据
    time.sleep(0.1)

关键说明 #

• 数据缓存：用data_buffer列表暂存每次读取的AcY值，避免单条发送的低效问题。
• 双条件触发：同时判断数据量和时间，既保证凑够100条就发送，也防止因传感器读取异常导致长时间无数据发布。
• JSON格式打包：将列表转为JSON字符串发送，Node-RED可通过json节点直接解析为数组，便于后续拆分入库。
• 读取频率调整：把原代码的1秒间隔改为0.1秒，确保10秒内能够收集到足够的批量数据，你可根据实际硬件性能微调该值。

Node-RED处理提示 #

在Node-RED中可按以下流程处理批量数据：

• 用mqtt in节点订阅目标主题
• 用json节点将收到的JSON字符串解析为数组
• 用split节点将数组拆分为单个数据项（如需逐条入库）
• 用对应数据库节点（如mysql/postgres）将数据写入数据库

内容的提问来源于stack exchange，提问作者ez.oez