# MQTT Streams EMQX 6.1では、MQTTのリアルタイムなパブリッシュ／サブスクライブモデルを拡張し、永続的でリプレイ可能なメッセージストリームを提供するストリーミングおよびリプレイ機能であるMQTT Streamsを導入しました。これにより、MQTTのセマンティクスを保持しつつ、Kafkaのようなストリーミング機能を実現します。 本ページでは、EMQXにおけるMQTT Streams機能の設計動機、主要概念、内部アーキテクチャ、メッセージフロー、実際の適用シナリオについて包括的に解説します。 ::: warning Listener Mountpointとの非互換性について [mountpoint](../../guides/configuration/listener.md#mountpoint)が設定されたリスナー経由で接続したクライアントには、ストリームのリプレイは機能しません。EMQXは`$stream/`プレフィックスのマッチング前にmountpointを適用するため、`$stream/`へのサブスクライブはマウントされたリテラルトピックへの通常のサブスクライブとして扱われ、クライアントにはエラーは通知されません。メッセージの取り込みも影響を受け、パブリッシュされたトピックにはmountpointプレフィックスが付与されるため、そのプレフィックスを含むトピックフィルターを持つストリームにのみマッチします。 ::: ## MQTT Streamとは何か？ MQTT Streamは、名前付きの論理リソースであり、設定されたトピックフィルターにマッチするMQTTメッセージを継続的に収集します。メッセージはストリームの保持ポリシーに従って永続的に保存され、後からサブスクライブしたクライアントによってリプレイ可能です。 ストリームは一意の名前で識別されます。トピックフィルターはストリームの設定の一部ですが、識別子ではありません。 各ストリームには以下が含まれます： - 一意の名前 - 設定されたトピックフィルター - 保持ポリシー（時間ベースおよび／またはサイズベース） - オプションのラストバリューセマンティクス - 明示的なライフサイクル（作成、更新、削除） ## なぜMQTT Streamsを使うのか？ MQTTはリアルタイムメッセージングに最適化されていますが、以下のような制約があります： - メッセージは通常オンラインのサブスクライバーにのみ配信される。 - 過去のデータのリプレイはネイティブにサポートされていない。 - 過去データの再処理には外部システムが必要。 - 順序付けられたリプレイ可能なメッセージ履歴の維持が困難。 MQTT Streamsは、永続的なメッセージ保存とリプレイをMQTTに拡張します。これにより、MQTTクライアントのパブリッシュやサブスクライブ方法を変えずに、過去のメッセージを読み取ったり、デバイスの最新状態を取得したりできます。 ## MQTT Streamsの主要概念 - **MQTT Stream** 名前で識別・アドレス指定され、明示的なライフサイクルで管理される論理リソースです。アクティブな間は、設定された時間またはサイズの制限内でマッチするメッセージを継続的に保存します。保存されたメッセージは、パブリッシャー側の変更なしにサブスクライバーがリプレイ可能です。 ストリーム名に使用できる文字は以下のみです： - 英数字（`A–Z`、`a–z`、`0–9`） - アンダースコア（`_`） - ハイフン（`-`） - ドット（`.`） サポートされるストリームタイプは2種類です： - **レギュラーストリーム**：過去データを上書きせずにすべてのマッチするメッセージを保存します。`stream-offset`サブスクリプションプロパティを使い、指定したタイムスタンプやオフセットからメッセージをリプレイ可能です。 - **ラストバリューストリーム**：[ラストバリューセマンティクス](#last-value-semantics)を有効にします。同じストリームキーのメッセージは新しいものが古いものを上書きし、各キーに関連付けられた最新のメッセージのみを保持します。詳細は[ストリームキー式](./mqtt-stream-task.md#stream-key-expression)を参照してください。 - **トピックフィルター** `sensors/+/data`のようなMQTTトピックフィルターで、どのパブリッシュメッセージをストリームに取り込むかを決定します。マッチするメッセージのみが取り込まれ、1つのメッセージが複数のストリームに属することもあります。 ::: tip トピックフィルターはストリームの識別子ではなく、名前付きストリームの設定メタデータです。 ::: - **ストリームサブスクリプション** ストリームからメッセージを消費するための特別なMQTTサブスクリプションです。クライアントは以下のいずれかの形式でサブスクライブします： ``` SUBSCRIBE $stream/ SUBSCRIBE $stream// ``` ここで： - `` はストリーム名（必須） - `` は既存のストリームにサブスクライブする場合は省略可能 - オートクリエーションが有効な場合、`$stream//`はストリームが存在しなければ指定されたトピックフィルターでストリームを作成します。 ストリームサブスクリプションは通常のMQTTサブスクリプションとは独立して動作し、External Subscription機構を通じて配信されます。 - **ストリームオフセット（リプレイ開始位置）** リプレイの開始位置は、トピックパスではなくMQTT 5のユーザーサブスクリプションプロパティ`stream-offset`で指定します。 `stream-offset`はリプレイ開始位置を決定します。例： - タイムスタンプ - 論理オフセット - 最も早い位置や最新位置などの特別な位置（サポートされている場合） この設計により、トピック文字列からオフセット解析を排除し、MQTT 5プロパティによるリプレイ制御と整合します。 - **キー式** 受信する各メッセージに対して評価されるユーザー定義式で、キーを抽出します。式はメッセージの内容やメタデータを参照可能です。抽出されたキーはストレージパーティション内のメッセージ順序保証に使われます。ラストバリューセマンティクスが有効な場合、キーは上書き範囲を定義し、同じキーの新しいメッセージが古いものを置き換えます。 ## MQTT Streamsのアーキテクチャ MQTT Streamsは、ブローカーコアと疎結合で動作し、既存のインフラを再利用する独立したEMQXアプリケーションとして実装されています。EMQXとの統合は内部フックとExternal Subscriptionフレームワークを通じて行われます。 ::: tip External Subscriptionは、ライブMQTTパブリッシュ経路外から発生するメッセージソースをMQTTクライアントセッションに接続し、クライアントの動作を変えずに標準MQTTサブスクリプションを通じてメッセージを配信するEMQXの仕組みです。 ::: ### 主なコンポーネント - **Streams Registry**：MQTTストリームのライフサイクルを管理し、ストリーム名、トピックフィルター、保持ポリシー、キー式などのメタデータを保持します。効率的なストリーム検索のためにMnesiaテーブルを使用します。 - **Streams Message Database**：ストリームメッセージの永続的ストレージを提供し、EMQXの[パーシステンス](../design/durable-storage.md)上に構築されています。メッセージを永続化し、保持制限を適用し、ラストバリューセマンティクスを有効にした場合は適用し、保持ポリシーに従って効率的なメッセージ取得をサポートします。 - **Streams ExtSub Handler**：メッセージストリームをMQTTクライアントセッションに統合します。パーシステントストレージからメッセージを取得し、External Subscriptionフレームワークを通じてサブスクライブクライアントに配信します。 ### MQTT Streamsのデータフローダイアグラム 以下の図はMQTT Streamsコンポーネント間のデータフローを示しています： ![streams_data_flow](./assets/mqtt_streams_data_flow.png) ### パブリッシュフロー 1. クライアントがMQTTトピックにメッセージをパブリッシュします。 2. MQTT Streamsフックがトリガーされてパブリッシュ処理を開始します。 3. フックはStreams Registryに問い合わせて、メッセージトピックにマッチするストリームを特定します。 4. マッチした各ストリームに対してメッセージを書き込み、パーシステントストレージに永続化します。 ### サブスクライブおよび消費フロー 1. クライアントがストリームトピック（`$stream/`または`$stream//`）にサブスクライブし、必要に応じて`stream-offset`サブスクリプションプロパティを含めます。 ::: warning 廃止予定 旧形式の`$s//`は後方互換性のためサポートされていますが廃止予定です。詳細は[互換性の注意点](#compatibility-notes)を参照してください。 ::: 2. External Subscriptionフレームワークがサブスクリプションを処理し、ストリームトピック用のStreams ExtSubハンドラーを初期化します。 3. ハンドラーは指定された`stream-offset`および保持ルールに従い、パーシステントストレージからメッセージを取得します。 4. 取得したメッセージはExternal Subscriptionフレームワークに渡されます。 5. ExtSubアプリケーションは標準MQTT配信を通じてクライアントにメッセージを届けます。 ## MQTT Streamsのコア機能 MQTT Streamsは、メッセージの保存、順序付け、保持、リプレイ消費のための配信方法を定義する一連のコア機能を提供します。 - **オフセットベースのリプレイ** リプレイ開始位置はトピックパスではなく`stream-offset`サブスクリプションプロパティで指定します。指定されたオフセット以前にパブリッシュされたメッセージはスキップされます。 - **保持** ストリームの保持ポリシーはメッセージのリプレイ範囲を制限します。メッセージは時間またはサイズの制限内で保持され、期限切れのメッセージは消費済みか否かに関わらず自動的に削除されます。 - **キー単位の順序保証** MQTT Streamsは単一のグローバルな配信順序を保証しません。同じキーを持つメッセージは常にパブリッシュ順に配信されます。異なるキーのメッセージは任意の順序で配信される可能性があります。 - **ラストバリューセマンティクス** ストリームはラストバリューセマンティクスを有効にできます。同じキーのメッセージは古いものを上書きし、最新の値のみを保持します。キーが解決できないメッセージは通常通り保存されます。 - **MQTTネイティブ配信** ストリームメッセージは標準MQTTメカニズムを用いて配信されます。パブリッシャーは動作を変更する必要がありません。サブスクライバーへのメッセージ配信はExternal Subscriptionを通じて統合されます。 ## セキュリティ上の考慮事項 EMQXは、`$stream/`プレフィックスおよびストリーム名を含む完全なサブスクリプショントピックフィルターに対してストリームサブスクリプションの認可を行います。ストリームサブスクリプションを認可すると、クライアントはサブスクリプション作成前にストリームが保存したメッセージをリプレイできる可能性があります。 ### ストリームサブスクリプションには専用のルールが必要 通常のトピックスペース用に書かれたルールは対応するストリームサブスクリプションをカバーしません： - `t/#`用のルールは`$stream/events/t/#`には適用されません。EMQXはこれらを異なるトピックとして扱います。 - `#`や`+`で始まる認可トピックフィルターは、`$`で始まるサブスクリプショントピックフィルターにマッチしません。デフォルトの`acl.conf`にある`{eq, "#"}`ルールを含む`#`拒否ルールは`$stream/events/#`を拒否しません。 したがって、`#`が拒否されたクライアントでも`$stream/events/#`にサブスクライブし、`stream-offset`を`earliest`に設定してストリームにまだ保存されているすべてのメッセージをリプレイできます。`$stream/`用に明示的なルールを追加し、ストリームのトピックフィルターにマッチするトピックのルールと同等以上に厳格にしてください： ```erlang %% 自動作成を許可せず、事前作成済みの"events"ストリームのリプレイを1つのコンシューマに許可。 {allow, {username, "event_reader"}, subscribe, ["$stream/events"]}. %% 廃止予定のプレフィックスを含むその他すべてのストリームサブスクリプションを拒否。 {deny, all, subscribe, ["$stream/#", "$s/#"]}. ``` クライアントがまだ使用している限り、廃止予定の`$s/`プレフィックスはルールに残してください。[廃止予定プレフィックス](#deprecated-prefix)を参照。 この動作は[共有サブスクリプション](../../get-started/messaging/mqtt-shared-subscription.md)とは異なります。`$share//t/#`の場合、EMQXはプレフィックスを除去して` t/#`を認可しますが、`$stream//t/#`の場合は完全なサブスクリプショントピックフィルターを認可します。 ### オートクリエーションはクライアント指定のトピックフィルターを許可 ストリームのオートクリエーションが有効な場合、サブスクライブするクライアントが新規ストリームのトピックフィルターを決定します。EMQXは`$stream//`の``を使ってストリームを作成し、クライアントが直接``へのサブスクライブ権限を持つかどうかは別途チェックしません。 例えば、`$stream/+/#`へのサブスクライブが許可されたクライアントは`$stream/events/#`にサブスクライブできます。`events`ストリームが存在しなければ、EMQXは`#`をトピックフィルターとしてストリームを作成します。これにより、`$`で始まらないすべてのトピックのメッセージを保存します。クライアントは直接サブスクライブ権限のないトピックのメッセージをリプレイできる可能性があります。 オートクリエーションはデフォルトでラストバリューストリームに対して有効、レギュラーストリームに対して無効ですが、MQTT Streamsが有効な場合にのみ有効です。MQTT Streamsはデフォルトで無効（`streams.enable = false`）です。信頼できないクライアントを受け入れる環境では、`$stream/events`など特定の事前作成済みストリームへのアクセスのみ許可し、`$stream/#`や`$s/#`にマッチするその他のサブスクリプションはすべて拒否してください。あるいはオートクリエーションを無効にして、ダッシュボードやREST APIからストリームを作成してください。[ダッシュボード経由でのストリーム自動作成](./mqtt-stream-task.md#automatically-create-streams-via-dashboard)を参照。 ## 互換性の注意点 既存のデプロイメントに関する互換性の考慮事項を説明します。 ### 名前付きストリーム - すべてのストリームは明示的に名前付きリソースとなりました。 - ストリーム名は許可された文字セットのルールに従う必要があります。 ### 旧ストリーム 以前に作成された名前なしストリームには、トピックフィルターから派生した名前が自動的に割り当てられます。 派生名は`/`となります。 ### 廃止予定プレフィックス ストリームサブスクライブ用の`$s`プレフィックスは後方互換性のために引き続きサポートされていますが、廃止予定です。 新規デプロイメントでは`$stream/`を使用してください。 ## 典型的なユースケース - **過去データのリプレイ**：過去のMQTTイベントをデバッグや新しいビジネスロジックのために再処理。 - **時系列分析**：センサーデータを保存・リプレイして分析や予知保全に活用。 - **イベントソーシング**：すべての状態変化を不変のイベントログとして永続化。 - **IoTデジタルツイン**：物理デバイスの最新状態をデジタルで保持。 - **設定同期**：デバイスが常に最新の設定を受け取ることを保証。 ## 次のステップ MQTT Streamsの基本を理解したら、実践方法を学びましょう： - [ストリームの作成と設定](./mqtt-stream-task.md)：ダッシュボードやREST APIを使ったストリームの宣言、ラストバリューセマンティクスや保持ポリシーの設定方法を学べます。 - [クイックスタートチュートリアル](./mqtt-stream-quick-start.md)：MQTTXを使った実際のパブリッシャー／サブスクライバーシナリオをステップバイステップで体験できます。