Posted inプログラミング

NestJSマイクロサービスとMQTTを使用したスケーラブルなIoTバックエンドの構築

Programming tutorial - IT technology blog
Programming tutorial - IT technology blog

背景と理由：IoTにおけるHTTPからの脱却

キャリアの初期、私は多くのウェブ開発者と同じように、REST APIを使用してIoTバックエンドを構築していました。少数のデバイスであれば問題なく動作しましたが、デバイス数が100台を超えたあたりで限界が見えてきました。HTTPヘッダーは重すぎたのです。数千のセンサーが数秒ごとに50バイトのペイロードを送信する場合、500バイトのヘッダーは帯域幅とバッテリー寿命の膨大な無駄になります。

MQTT (Message Queuing Telemetry Transport) は、軽量な接続を維持することでこの問題を解決します。これが業界標準であるのには理由があります。生のMQTTはすぐに管理不能なスパゲッティコードになりがちですが、NestJSはマイクロサービスの抽象化を提供し、コードをクリーンに保ちます。NestJSを使用することで、MQTTメッセージを標準的なイベントとして扱うことができ、データの取り込みをコアのビジネスロジックから切り離すことが可能になります。

最近のプロジェクトでこのアーキテクチャに切り替えたところ、サーバーのCPU使用率が40%削減されました。ファームウェアの配布中にデバイスのチェックインが急増しても、システムは安定していました。このガイドでは、高頻度のセンサーデータを無理なく処理するために私が実際に使用しているセットアップ手順を詳しく解説します。

インストール：環境のセットアップ

データを処理する前にメッセージブローカーが必要です。大規模なスケールにはEMQXが適していますが、開発や中規模のワークロードにはEclipse Mosquittoがおすすめです。非常に軽量で、低負荷時には10MB未満のRAMしか消費しないこともよくあります。

1. DockerでMosquittoを実行する

Dockerは、ローカル環境を汚さずにブローカーを起動する最速の方法です。docker-compose.ymlファイルを作成してサービスを立ち上げましょう。

version: '3.8'
services:
  mosquitto:
    image: eclipse-mosquitto
    container_name: mosquitto_broker
    ports:
      - "1883:1883" # 標準MQTTポート
      - "9001:9001" # WebSocketsポート
    volumes:
      - ./mosquitto.conf:/mosquitto/config/mosquitto.conf

テストを簡単にするため、匿名接続を許可する基本的なmosquitto.confを使用します。

persistence true
allow_anonymous true
listener 1883 0.0.0.0

2. NestJSプロジェクトの初期化

まず、NestJS CLIをインストールします。システムをGatewayとTelemetry Serviceに分割することをお勧めしますが、このチュートリアルではコアとなるマイクロサービスのセットアップに焦点を当てます。

# NestJS CLIをインストール
npm install -g @nestjs/cli

# プロジェクトを作成
nest new iot-backend
cd iot-backend

# マイクロサービスとMQTTトランスポートパッケージをインストール
npm install @nestjs/microservices mqtt

設定：マイクロサービスロジックの実装

NestJSはMQTTをトランスポートレイヤーとして扱います。@MessagePatternや@EventPatternなどのデコレータを使用して着信データを処理できます。テレメトリの場合、@EventPatternの方が適しています。これは「送りっぱなし（fire-and-forget）」で動作するため、高頻度のセンサー更新に最適だからです。

1. マイクロサービスのエントリポイントの設定

main.tsを修正して、標準のHTTPリクエストではなくMQTTメッセージをリッスンするようにNestJSに指示します。これにより、アプリケーションが専用のメッセージプロセッサに変換されます。

import { NestFactory } from '@nestjs/core';
import { Transport, MicroserviceOptions } from '@nestjs/microservices';
import { AppModule } from './app.module';

async function bootstrap() {
  const app = await NestFactory.createMicroservice<MicroserviceOptions>(
    AppModule,
    {
      transport: Transport.MQTT,
      options: {
        url: 'mqtt://localhost:1883',
        // 本番環境では、セッション状態を追跡するために一意のclientIdを使用してください
      },
    },
  );
  await app.listen();
  console.log('テレメトリマイクロサービスが起動しました');
}
bootstrap();

2. 着信センサーデータの処理

次に、特定のトピックをリッスンするコントローラーを作成します。+ワイルドカードを使用して、任意のセンサーIDにマッチさせます。例えば、sensors/sensor_01/dataとsensors/sensor_02/dataはどちらも同じハンドラーをトリガーします。

import { Controller } from '@nestjs/common';
import { EventPattern, Payload, Ctx, MqttContext } from '@nestjs/microservices';

@Controller()
export class TelemetryController {
  
  @EventPattern('sensors/+/data')
  handleTelemetry(@Payload() data: any, @Ctx() context: MqttContext) {
    const topic = context.getTopic();
    const sensorId = topic.split('/')[1];
    
    // ここでデータベースへの挿入など、10〜20ミリ秒程度のタスクを処理します
    console.log(`センサー [${sensorId}] を処理中:`, data);
    this.processData(sensorId, data);
  }

  private processData(id: string, payload: any) {
    // ここにビジネスロジックを記述します
  }
}

3. デバイスへのコマンド送信

IoTにおける通信は双方向です。ディスプレイの更新やリレーの切り替えなど、ハードウェアにコマンドをプッシュする必要が頻繁にあります。ClientProxyを使用すれば、これを簡単に行えます。

import { Injectable, Inject } from '@nestjs/common';
import { ClientProxy } from '@nestjs/microservices';

@Injectable()
export class CommandService {
  constructor(
    @Inject('MQTT_SERVICE') private client: ClientProxy,
  ) {}

  sendCommand(deviceId: string, command: string) {
    const pattern = `devices/${deviceId}/commands`;
    const payload = { action: command, timestamp: new Date().toISOString() };
    
    return this.client.emit(pattern, payload);
  }
}

検証とスケーリング：本番環境での考慮事項

コードを書くことは戦いの半分に過ぎません。システムが実際のトラフィックや接続断に耐えられるかを確認する必要があります。

1. MQTT Explorerによる可視化

メッセージが送信されているかどうかを推測するのはやめましょう。MQTT Explorerをダウンロードしてlocalhost:1883に接続してください。トピックツリーをリアルタイムで可視化できます。NestJSサービスが反応しない場合は、まずこのツールを使ってブローカーが実際にパケットを受信しているか確認してください。

2. 水平スケーリングと共有サブスクリプション

MQTTの大きなハードルの一つは、サービスの複数のインスタンスがそれぞれ同じメッセージのコピーを受信してしまうことです。これはデータベースの重複エントリにつながります。これを解決するには、共有サブスクリプション (Shared Subscriptions)を使用します。トピックの先頭に$share/group_name/を付けることで、ブローカーはすべてのアクティブなサービスインスタンス間でメッセージをロードバランシングします。これにより、秒間数十万メッセージまでスケールさせることが可能になります。

3. 適切なQoSの選択

MQTTには3つのサービス品質 (QoS) レベルがあります。ほとんどのテレメトリでは、QoS 1 (少なくとも1回) が最適です。メッセージがバックエンドに到達することを保証しつつ、オーバーヘッドを最小限に抑えられます。NestJSは、コントローラー関数の実行が完了すると自動的に確認応答 (PUBACK) を送信します。処理の途中でコードがクラッシュした場合、ブローカーは配信を再試行するため、再起動中に重要なデータが失われることはありません。

このアーキテクチャにより、関心の分離が明確になります。開発チームはソケットのハートビート管理ではなく、データの処理に集中できるようになります。長期的なIoTプロジェクトにおいて、NestJSとMQTTの組み合わせは、デバイス数の増加に合わせて拡張できる堅牢な基盤となります。

Share:
Tags: