ESP32到Web：实时IoT数据管道（MQTT + Node.js + WebSockets） | WellAlly康心伴

# 拉取Mosquitto镜像
docker pull eclipse-mosquitto

# 运行broker
docker run -it -p 1883:1883 -p 9001:9001 -v $(pwd)/mosquitto.conf:/mosquitto/config/mosquitto.conf eclipse-mosquitto

mkdir wearable-pipeline-backend
cd wearable-pipeline-backend
npm init -y

npm install mqtt ws

{
  "name": "wearable-pipeline-backend",
  "version": "1.0.0",
  "description": "",
  "main": "server.js",
  "scripts": {
    "start": "node server.js"
  },
  "keywords": [],
  "author": "",
  "license": "ISC",
  "dependencies": {
    "mqtt": "^5.3.5",
    "ws": "^8.16.0"
  }
}

// src/esp32_mqtt_publisher/esp32_mqtt_publisher.ino

#include <WiFi.h>
#include <PubSubClient.h>
#include <Adafruit_BME280.h>
#include <ArduinoJson.h>

// --- Wi-Fi和MQTT配置 ---
const char* ssid = "YOUR_WIFI_SSID";
const char* password = "YOUR_WIFI_PASSWORD";
const char* mqtt_server = "broker.emqx.io";
const int mqtt_port = 1883;

// --- MQTT主题 ---
const char* temp_topic = "wearable/device1/temperature";
const char* humidity_topic = "wearable/device1/humidity";

// --- 全局变量 ---
WiFiClient espClient;
PubSubClient client(espClient);
Adafruit_BME280 bme; // I2C
unsigned long lastMsg = 0;
#define MSG_BUFFER_SIZE (50)
char msg[MSG_BUFFER_SIZE];

void setup_wifi() {
  delay(10);
  Serial.println();
  Serial.print("正在连接到 ");
  Serial.println(ssid);
  WiFi.begin(ssid, password);
  while (WiFi.status() != WL_CONNECTED) {
    delay(500);
    Serial.print(".");
  }
  Serial.println("");
  Serial.println("WiFi已连接");
  Serial.println("IP地址: ");
  Serial.println(WiFi.localIP());
}

void reconnect() {
  // 循环直到重新连接
  while (!client.connected()) {
    Serial.print("正在尝试MQTT连接...");
    // 创建随机客户端ID
    String clientId = "ESP32Client-";
    clientId += String(random(0xffff), HEX);
    // 尝试连接
    if (client.connect(clientId.c_str())) {
      Serial.println("已连接");
    } else {
      Serial.print("失败, rc=");
      Serial.print(client.state());
      Serial.println(" 5秒后重试");
      // 等待5秒后重试
      delay(5000);
    }
  }
}

void setup() {
  Serial.begin(115200);
  setup_wifi();
  client.setServer(mqtt_server, mqtt_port);

  if (!bme.begin(0x76)) {
    Serial.println("找不到有效的BME280传感器，请检查接线！");
    while (1);
  }
}

void loop() {
  if (!client.connected()) {
    reconnect();
  }
  client.loop();

  unsigned long now = millis();
  if (now - lastMsg > 5000) { // 每5秒发布一次
    lastMsg = now;

    float temperature = bme.readTemperature(); // 摄氏度
    float humidity = bme.readHumidity();

    if (isnan(temperature) || isnan(humidity)) {
        Serial.println("从BME传感器读取失败！");
        return;
    }

    // 准备温度的JSON负载
    StaticJsonDocument＜128> tempDoc;
    tempDoc["deviceId"] = "device1";
    tempDoc["value"] = temperature;
    tempDoc["unit"] = "C";
    tempDoc["timestamp"] = millis();
    char tempOutput[128];
    serializeJson(tempDoc, tempOutput);

    // 发布温度
    client.publish(temp_topic, tempOutput);
    Serial.print("已发布到 ");
    Serial.print(temp_topic);
    Serial.println(tempOutput);

    // 准备湿度的JSON负载
    StaticJsonDocument＜128> humidityDoc;
    humidityDoc["deviceId"] = "device1";
    humidityDoc["value"] = humidity;
    humidityDoc["unit"] = "%";
    humidityDoc["timestamp"] = millis();
    char humidityOutput[128];
    serializeJson(humidityDoc, humidityOutput);

    // 发布湿度
    client.publish(humidity_topic, humidityOutput);
    Serial.print("已发布到 ");
    Serial.print(humidity_topic);
    Serial.println(humidityOutput);
  }
}

// server.js

const mqtt = require('mqtt');
const { WebSocketServer } = require('ws');

// --- MQTT配置 ---
const MQTT_BROKER = 'mqtt://broker.emqx.io';
const TOPICS = ['wearable/device1/temperature', 'wearable/device1/humidity'];

// --- WebSocket服务器配置 ---
const WSS_PORT = 8080;

// 1. 创建WebSocket服务器
const wss = new WebSocketServer({ port: WSS_PORT });
console.log(`WebSocket服务器已在端口 ${WSS_PORT} 启动`);

wss.on('connection', (ws) => {
  console.log('客户端已连接到WebSocket');
  ws.on('close', () => console.log('客户端已断开'));
});

// 广播到所有客户端的函数
function broadcast(topic, message) {
  console.log(`正在广播给 ${wss.clients.size} 个客户端:`, topic, message.toString());
  wss.clients.forEach(client => {
    if (client.readyState === 1) { // WebSocket.OPEN = 1
      client.send(JSON.stringify({ topic, data: JSON.parse(message.toString()) }));
    }
  });
}

// 2. 连接到MQTT Broker
const client = mqtt.connect(MQTT_BROKER);

client.on('connect', () => {
  console.log('已连接到MQTT broker');

  // 3. 订阅所有主题
  TOPICS.forEach(topic => {
    client.subscribe(topic, (err) => {
      if (!err) {
        console.log(`已订阅主题: ${topic}`);
      }
    });
  });
});

// 4. 处理传入消息
client.on('message', (topic, message) => {
  console.log(`收到来自主题 ${topic} 的消息: ${message.toString()}`);

  // 将消息广播到所有WebSocket客户端
  broadcast(topic, message);
});

client.on('error', (err) => {
    console.error('MQTT错误:', err);
    client.end();
});

node server.js

<!-- index.html -->
<!DOCTYPE html>
<html lang="zh-CN">
<head>
    <meta charset="UTF-8">
    <meta name="viewport" content="width=device-width, initial-scale=1.0">
    <title>实时可穿戴设备数据</title>
    <script src="https://cdn.jsdelivr.net/npm/chart.js"></script>
    <style>
        body { font-family: sans-serif; display: flex; justify-content: center; align-items: center; height: 100vh; background-color: #f0f2f5; }
        .dashboard { display: flex; gap: 20px; }
        .chart-container { width: 500px; padding: 20px; background: white; border-radius: 8px; box-shadow: 0 2px 4px rgba(0,0,0,0.1); }
        h1 { text-align: center; }
    </style>
</head>
<body>
    <div class="dashboard">
        <div class="chart-container">
            <h1>温度 (°C)</h1>
            <canvas id="tempChart"></canvas>
        </div>
        <div class="chart-container">
            <h1>湿度 (%)</h1>
            <canvas id="humidityChart"></canvas>
        </div>
    </div>

<script>
    const MAX_DATA_POINTS = 20;

    // --- 温度的Chart.js设置 ---
    const tempCtx = document.getElementById('tempChart').getContext('2d');
    const tempChart = new Chart(tempCtx, {
        type: 'line',
        data: {
            labels: [],
            datasets: [{
                label: '温度',
                data: [],
                borderColor: 'rgba(255, 99, 132, 1)',
                backgroundColor: 'rgba(255, 99, 132, 0.2)',
                borderWidth: 2,
                fill: true
            }]
        },
        options: {
            scales: {
                x: { title: { display: true, text: '时间' } },
                y: { title: { display: true, text: '数值' } }
            },
            animation: { duration: 200 }
        }
    });

    // --- 湿度的Chart.js设置 ---
    const humidityCtx = document.getElementById('humidityChart').getContext('2d');
    const humidityChart = new Chart(humidityCtx, {
        type: 'line',
        data: {
            labels: [],
            datasets: [{
                label: '湿度',
                data: [],
                borderColor: 'rgba(54, 162, 235, 1)',
                backgroundColor: 'rgba(54, 162, 235, 0.2)',
                borderWidth: 2,
                fill: true
            }]
        },
        options: {
            scales: {
                x: { title: { display: true, text: '时间' } },
                y: { title: { display: true, text: '数值' } }
            },
            animation: { duration: 200 }
        }
    });

    function addData(chart, label, data) {
        chart.data.labels.push(label);
        chart.data.datasets.forEach((dataset) => {
            dataset.data.push(data);
        });

        // 限制数据点数量
        if (chart.data.labels.length > MAX_DATA_POINTS) {
            chart.data.labels.shift();
            chart.data.datasets.forEach((dataset) => {
                dataset.data.shift();
            });
        }
        chart.update();
    }

    // --- WebSocket连接 ---
    const socket = new WebSocket('ws://localhost:8080');

    socket.onopen = () => {
        console.log('WebSocket连接已建立');
    };

    socket.onmessage = (event) => {
        const message = JSON.parse(event.data);
        const topic = message.topic;
        const data = message.data;
        const now = new Date();
        const timeLabel = `${now.getHours()}:${now.getMinutes()}:${now.getSeconds()}`;

        console.log(`收到来自 ${topic}:`, data);

        if (topic.includes('temperature')) {
            addData(tempChart, timeLabel, data.value);
        } else if (topic.includes('humidity')) {
            addData(humidityChart, timeLabel, data.value);
        }
    };

    socket.onclose = () => {
        console.log('WebSocket连接已关闭');
    };

    socket.onerror = (error) => {
        console.error('WebSocket错误:', error);
    };

</script>
</body>
</html>