ウェアラブルからWebへ：IoT、MQTT、Node.jsによるリアルタイムデータパイプライン

完全なリアルタイムIoTデータパイプラインを構築する詳細なチュートリアル。ESP32ウェアラブルのセンサーデータをMQTTブローカーを通じてNode.jsバックエンドにストリーミングし、WebSocketを使用してWebアプリ上でリアルタイムに可視化します。

WellAlly編集チーム

著者プロフィールを見る

2025年12月12日読了時間 10分

このページの内容

はじめに

フィットネストラッカーがリアルタイムで心拍数をスマホに送信する仕組みやスマートホームのセンサーがダッシュボードを即座に更新する仕組みに驚いたことはありますか？その魔法は、堅牢なリアルタイムデータパイプラインにあります。開発者にとって、これらのパイプラインを構築することは、急速に成長しているIoT（Internet of Things）の世界におけるコアスキルです。

この深度チュートリアルでは、完全なエンドツーエンドのデータパイプラインを構築します。人気があり手頃な価格のESP32マイコンを使ってウェアラブルデバイスをシミュレートし、軽量なMQTTプロトコルでセンサーデータをストリーミングし、Node.jsバックエンドで処理し、WebSocketを使ってウェブダッシュボードでライブ可視化を行います。

このプロジェクトは単なる理論的な演習ではなく、パーソナルヘルスモニターから産業用センサーネットワークまで、独自のIoTアプリケーションを構築するための設計図です。リアルタイムデータを効率的に処理し、低レイテンシーと高信頼性を確保する方法を学びます。

前提条件：

JavaScript（Node.js）とC++（Arduino）の基本的な理解。
おマシンにNode.jsとnpmがインストールされていること。
ESP32開発用にArduino IDEが設定されていること。
Wi-Fi搭載のESP32開発ボード（ любой モデルで可）。
ESP32用のセンサー（温度と湿度にBME280を使用しますが、任意のセンサーに簡単に適応可能）。

問題の理解：IoTデータ課題

何百万ものデバイスからデータをストリーミングすることは複雑です。各デバイスから中央サーバーへHTTPリクエストを送信する単純なアプローチは拡張性がありません。非効率的で、密結合になり、信頼性の低いネットワークに苦労します—IoTデバイスに共通のシナリオ。

ここで、**出版/購読者（pub/sub）**モデル、特にMQTTプロトコルが活躍します。

疎結合：デバイス（出版者）は、誰が受信するかを知らずに、特定の「トピック」でブローカーと呼ばれる中央ハブにメッセージを送信します。アプリケーション（購読者）は、必要なデータを取得するためにこれらのトピックをリッスンします。
軽量：MQTTは制約のあるデバイスと低帯域幅ネットワーク向けに設計されており、バッテリー駆動のウェアラブルに最適です。
信頼性：MQTTには、信頼性の低い接続でもメッセージ配信を保証するサービス品質（QoS）レベルが組み込まれています。

アーキテクチャは次のようになります：

ESP32（ウェアラブル） -> MQTTブローカー -> Node.jsバックエンド -> WebSocket -> ウェブブラウザー

このアプローチは拡張性があり、レジリエントであり、多くのIoTアプリケーションの業界標準です。

前提条件とセットアップ

コーディングを開始する前に、環境を準備しましょう。

1. MQTTブローカー

ブローカーは中央メッセージハブです。このチュートリアルには2つの優れたオプションがあります：

最も簡単（クラウドベース）：EMQ Xの бесплатный、公的なMQTTブローカーをbroker.emqx.ioで使用します。設定が不要で、快速開発に最適です。
ローカルで安全：より本番環境のような設定では、独自のMosquittoブローカーをインストールできます。Dockerで簡単に行えます：

bash

# Pull the Mosquitto image
docker pull eclipse-mosquitto

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

（このチュートリアルでは、シンプルさのためにすべての例で公的なbroker.emqx.ioを使用します。）

2. Node.jsプロジェクトセットアップ

コンピュータで、新しいプロジェクトディレクトリを作成し、Node.jsプロジェクトを初期化します。

bash

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

次に、必要なライブラリをインストールします：ブローカー接続用のmqttとWebSocketサーバー作成用のws。

bash

npm install mqtt ws

package.jsonは次のようになります：

json

{
  "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"
  }
}

3. ESP32用のArduino IDEセットアップ

Arduino IDEがESP32ボード用に設定されていることを確認してください。MQTT用のPubSubClientライブラリと、センサー用の必要なライブラリ（例：Adafruit BME280ライブラリ）も必要です。

Arduino IDEのライブラリマネージャーからこれらをインストールできます：

**スケッチ -> ライブラリをインクルード -> ライブラリを管理...**に移動します
PubSubClientを検索してインストールします。
Adafruit BME280 Libraryとその依存関係であるAdafruit Unified Sensorを検索してインストールします。

ステップ1：ウェアラブルデバイス（ESP32）

做什么

ESP32をプログラムして、センサーデータを定期的に読み取り、特定のトピックでMQTTブローカーにJSONペイロードとしてパブリッシュします。

実装

Arduino IDEを起動し、以下のコードを貼り付けます。Wi-Fiの認証情報を自分のものに変更することを忘れないでください！

cpp

// src/esp32_mqtt_publisher/esp32_mqtt_publisher.ino

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

// --- Wi-Fi and MQTT Configuration ---
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 Topics ---
const char* temp_topic = "wearable/device1/temperature";
const char* humidity_topic = "wearable/device1/humidity";

// --- Globals ---
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("Connecting to ");
  Serial.println(ssid);
  WiFi.begin(ssid, password);
  while (WiFi.status() != WL_CONNECTED) {
    delay(500);
    Serial.print(".");
  }
  Serial.println("");
  Serial.println("WiFi connected");
  Serial.println("IP address: ");
  Serial.println(WiFi.localIP());
}

void reconnect() {
  // Loop until we're reconnected
  while (!client.connected()) {
    Serial.print("Attempting MQTT connection...");
    // Create a random client ID
    String clientId = "ESP32Client-";
    clientId += String(random(0xffff), HEX);
    // Attempt to connect
    if (client.connect(clientId.c_str())) {
      Serial.println("connected");
    } else {
      Serial.print("failed, rc=");
      Serial.print(client.state());
      Serial.println(" try again in 5 seconds");
      // Wait 5 seconds before retrying
      delay(5000);
    }
  }
}

void setup() {
  Serial.begin(115200);
  setup_wifi();
  client.setServer(mqtt_server, mqtt_port);
  
  if (!bme.begin(0x76)) {
    Serial.println("Could not find a valid BME280 sensor, check wiring!");
    while (1);
  }
}

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

  unsigned long now = millis();
  if (now - lastMsg > 5000) { // Publish every 5 seconds
    lastMsg = now;

    float temperature = bme.readTemperature(); // Celsius
    float humidity = bme.readHumidity();

    if (isnan(temperature) || isnan(humidity)) {
        Serial.println("Failed to read from BME sensor!");
        return;
    }

    // Prepare JSON payload for temperature
    StaticJsonDocument<128> tempDoc;
    tempDoc["deviceId"] = "device1";
    tempDoc["value"] = temperature;
    tempDoc["unit"] = "C";
    tempDoc["timestamp"] = millis();
    char tempOutput[128];
    serializeJson(tempDoc, tempOutput);

    // Publish temperature
    client.publish(temp_topic, tempOutput);
    Serial.print("Published to ");
    Serial.print(temp_topic);
    Serial.println(tempOutput);
    
    // Prepare JSON payload for humidity
    StaticJsonDocument<128> humidityDoc;
    humidityDoc["deviceId"] = "device1";
    humidityDoc["value"] = humidity;
    humidityDoc["unit"] = "%";
    humidityDoc["timestamp"] = millis();
    char humidityOutput[128];
    serializeJson(humidityDoc, humidityOutput);

    // Publish humidity
    client.publish(humidity_topic, humidityOutput);
    Serial.print("Published to ");
    Serial.print(humidity_topic);
    Serial.println(humidityOutput);
  }
}

動作の仕組み

setup_wifi()：ESP32をローカルWi-Fiネットワークに接続します。
reconnect()：MQTTブローカーへの接続を処理する関数です。接続が切断された場合、再試行し続けます。これはデバイスのレジリエンスに重要です。
setup()：シリアルモニター、Wi-Fi、MQTTサーバー接続、BME280センサーを初期化します。
loop()：メインロジックです。MQTTクライアントが接続されていることを確認し、その後5秒ごとに温度と湿度を読み取ります。ArduinoJsonを使ってこのデータを単純なJSONオブジェクトにフォーマットし、client.publish()を使って各 Readings をそれぞれのトピック（wearable/device1/temperatureとwearable/device1/humidity）にパブリッシュします。

一般的な落とし穴

Wi-Fi認証情報が正しくない：SSIDとパスワードを二重確認してください。正しくない場合、シリアルモニターは「Connecting to...」で停止します。
ファイアウォール問題：ネットワークが標準MQTTポート1883をブロックしていないことを確認してください。
センサー配線：「Could not find a valid BME280 sensor」と表示された場合、I2C配線（SDA/SCLピン）を確認してください。

コードをアップロードしてシリアルモニターを開きます。Wi-Fiに接続し、データの publier を開始するのが見えるはずです！✨

ステップ2：Node.jsバックエンド（MQTT-to-WebSocketブリッジ）

做什么

これがパイプラインの核心です。以下のことを行うNode.jsサーバーを作成します：

MQTTブローカーに接続し、ESP32が publier しているトピックをサブスクライブします。
ウェブブラウザーからの接続を受け入れるWebSocketサーバーを作成します。
MQTTメッセージを受信すると、即座にそのメッセージをすべての接続されたWebSocketクライアントにブロードキャストします。

実装

wearable-pipeline-backendディレクトリに、server.jsという名前のファイルを作成します。

javascript

// server.js

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

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

// --- WebSocket Server Configuration ---
const WSS_PORT = 8080;

// 1. Create a WebSocket server
const wss = new WebSocketServer({ port: WSS_PORT });
console.log(`WebSocket server started on port ${WSS_PORT}`);

wss.on('connection', (ws) => {
  console.log('Client connected to WebSocket');
  ws.on('close', () => console.log('Client disconnected'));
});

// Function to broadcast to all clients
function broadcast(topic, message) {
  console.log(`Broadcasting to ${wss.clients.size} clients:`, 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. Connect to MQTT Broker
const client = mqtt.connect(MQTT_BROKER);

client.on('connect', () => {
  console.log('Connected to MQTT broker');
  
  // 3. Subscribe to all topics
  TOPICS.forEach(topic => {
    client.subscribe(topic, (err) => {
      if (!err) {
        console.log(`Subscribed to topic: ${topic}`);
      }
    });
  });
});

// 4. Handle incoming messages
client.on('message', (topic, message) => {
  console.log(`Received message from topic ${topic}: ${message.toString()}`);
  
  // Broadcast the message to all WebSocket clients
  broadcast(topic, message);
});

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

動作の仕組み

WebSocketサーバーセットアップ：ポート8080で新しいWebSocketServerインスタンスを作成します。クライアントが接続および切断的时候我々をログに記録するためにシンプルなログを追加します。
MQTTクライアントセットアップ：mqtt.connect()を使用して公的なブローカーに接続します。
サブスクリプション：client.on('connect', ...)コールバックで、TOPICS配列をループして、それぞれをサブスクライブします。これにより、ブローカーにこれらのトピックで publier されたメッセージを受け取りたいことを伝えます。
メッセージ処理：client.on('message', ...)イベントが魔法のリンクです。ブローカーからメッセージが届くと、この関数がトリガーされます。次に、broadcast関数を呼び出します。
ブロードキャスティング：broadcast関数はすべての接続されたWebSocketクライアント（wss.clients）を反復処理し、データを.send()します。トピックと解析されたメッセージデータをフロントエンドで fácil に処理するための新しいJSONオブジェクトにラップします。

ターミナルからサーバーを実行します：

bash

node server.js

MQTTブローカーに接続し、トピックをサブスクライブするのが見えるはずです。ESP32がまだ実行中の場合、ターミナルに到着するメッセージがログに記録されます！

ステップ3：ウェブアプリ（フロントエンド可視化）

做什么

最後に、ユーザー向けの部分を作成します：Node.js WebSocketサーバーに接続し、Chart.jsを使って着信データをリアルタイムで可視化するJavaScriptを含む単純なHTMLページです。

実装

バックエンドプロジェクトフォルダーに、index.htmlという名前の新しいファイルを作成します。

html

<!-- index.html -->
<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <meta name="viewport" content="width=device-width, initial-scale=1.0">
    <title>Real-Time Wearable Data</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>Temperature (°C)</h1>
            <canvas id="tempChart"></canvas>
        </div>
        <div class="chart-container">
            <h1>Humidity (%)</h1>
            <canvas id="humidityChart"></canvas>
        </div>
    </div>

<script>
    const MAX_DATA_POINTS = 20;

    // --- Chart.js Setup for Temperature ---
    const tempCtx = document.getElementById('tempChart').getContext('2d');
    const tempChart = new Chart(tempCtx, {
        type: 'line',
        data: {
            labels: [],
            datasets: [{
                label: 'Temperature',
                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: 'Time' } },
                y: { title: { display: true, text: 'Value' } }
            },
            animation: { duration: 200 }
        }
    });

    // --- Chart.js Setup for Humidity ---
    const humidityCtx = document.getElementById('humidityChart').getContext('2d');
    const humidityChart = new Chart(humidityCtx, {
        type: 'line',
        data: {
            labels: [],
            datasets: [{
                label: 'Humidity',
                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: 'Time' } },
                y: { title: { display: true, text: 'Value' } }
            },
            animation: { duration: 200 }
        }
    });

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

        // Limit the number of data points
        if (chart.data.labels.length > MAX_DATA_POINTS) {
            chart.data.labels.shift();
            chart.data.datasets.forEach((dataset) => {
                dataset.data.shift();
            });
        }
        chart.update();
    }

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

    socket.onopen = () => {
        console.log('WebSocket connection established');
    };

    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(`Received from ${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 connection closed');
    };

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

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

動作の仕組み

チャートセットアップ：Chart.jsライブラリを使用して、温度と湿度用に2つの折れ線グラフを作成します。基本的なスタイリングとラベルで構成します。
WebSocketクライアント：新しいWebSocketインスタンスを作成し、ws://localhost:8080でバックエンドサーバーに接続します。
socket.onmessage：Node.jsサーバーからプッシュされたデータを受け取る場所です。着信JSON文字列を解析します。
データルーティング：メッセージのtopicをチェックして、どちらのチャートを更新するかを決定します。
addData()関数：このヘルパー関数は、新しいデータポイントとタイムスタンプラベルを正しいチャートにプッシュします。チャートを読みやすく保つために、MAX_DATA_POINTSを超えると古いデータポイントも削除し、「スライディングウィンドウ」効果を作成します。

実際に見るには、index.htmlファイルをウェブブラウザで開くだけです。ESP32からデータが届くと、チャートがリアルタイムで更新されます！

セキュリティのベストプラクティス

現在の設定は開発には最適ですが、本番環境ではセキュリティは交渉不可能です。重要な次のステップは次のとおりです。

ブローカーでの匿名アクセスを無効にする：本番Mosquitto設定では、匿名の接続を無効にし、パスワードでユーザーアカウントを作成する必要があります。
- mosquitto_passwdを使用してパスワードファイルを作成します。
- mosquitto.confでallow_anonymous falseとpassword_file /path/to/passwordfileを設定します。
TLS/SSL暗号化を使用：MQTTとWebSocketトラフィックは暗号化される必要があります。
- MQTTブローカーをポート8883でTLSを使用するように設定します。ESP32とNode.jsクライアントは、安全な接続（mqtts://とwss://）を使用するように設定する必要があります。
- これにより、攻撃者がセンサーデータや認証情報をスニフできる man-in-the-middle 攻撃を防ぎます。
承認（ACL）の実装：最小権限の原則を強制するためにアクセス制御リスト（ACL）を使用します。
- 特定のデバイス（device1）は、自身トピック（wearable/device1/+）への publier のみを許可されるべきです。
- バックエンドサーバーは、必要なトピック（wearable/+/+）への publier のみを許可されるべきです。
- これにより、侵害されたデバイスが他のトピックに悪意のあるデータを publier したり、アクセス権のないデータをリッスンしたりすることを防ぎます。

結論

おめでとうございます！完全なリアルタイムIoTデータパイプラインを構築しました。ESP32の物理センサーから、拡張可能なMQTTブローカー、リアクティブなNode.jsバックエンド、そして finally、ウェブブラウザーでライブ更新ダッシュボードまで行いました。

これで、構築するための堅実な基盤があります。以下の方法でこのプロジェクトを拡張できます：

より多くのセンサーとデバイスを追加する。
時系列データベース（InfluxDBやTimescaleDBなど）にデータを保存して履歴分析を行う。
センサーバリューが特定のしきい値を超えたときに通知を送信するバックエンドにアラートルールを追加する。
React、Vue、Svelteでより精巧なフロントエンドダッシュボードを構築する。

MQTTとWebSocketを使用したpub/subアーキテクチャは非常に強力で、無数のプロフェッショナルなIoTシステムのバックボーンを形成しています。继续して構築してください！

参考資料

ESP32 MQTT：Random Nerd Tutorials - ESP32 MQTT Publish Subscribe
Node.js MQTT Client：mqtt.js on GitHub
Node.js WebSocket Library：ws on npm
Chart.js：Official Documentation
MQTT Security：HiveMQ MQTT Security Fundamentals

クイックリファレンス

側面	主要情報
トピック	開発
関連性	患者教育と認識
アクション	医療チームと話し合う
頻度	ケアチーム推奨通り
証拠	現在の臨床ガイドラインに基づく

よくある質問

Q: これが自分に該当するかどうを確認するには？ A: 個人的な病歴と家族歴について医療提供者と相談し、個人的なリスク要因と適切な次のステップを判断してください。

Q: 医師に何を尋ねるべきか？ A: 良い質問には：「結果は何を意味しますか？」「オプションは何ですか？」「どんな生活習慣の変化が有帮助ですか？」「いつフォローアップすべきですか？」があります。

Q: 帮助になる生活習慣の変更はありますか？ A: ほとんどの健康状態は、基本的な対策から benefits を受けます：バランスの取れた栄養、定期的な運動（週150分以上）、質の良い睡眠（7〜9時間）、ストレス管理、タバコの回避。

医療免責事項

この記事は教育目的のみを目的としており、医学的アドバイスではありません。健康、治療、ケアに関する決定する前に、常に資格のある医療提供者と相談してください。個人の状況は異なり、一人に適切なものが別の人にも適切であるとは限りません。

医学的緊急状況が発生した場合は、直ちに119番（または現地の緊急番号）に電話してください。

すべての記事に戻る